holm/django-helpdesk
1
0
Fork 0
forked from extern/django-helpdesk
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge branch 'main' into patch-1

Browse Source
This commit is contained in:
Christopher Broderick 2023-04-19 12:14:25 +01:00 committed by GitHub
commit 4c877bbe2f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 359 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

121
helpdesk/email.py
View File

@ -28,10 +28,12 @@ from helpdesk.models import FollowUp, IgnoreEmail, Queue, Ticket
import imaplib
import logging
import mimetypes
import oauthlib.oauth2 as oauth2lib
import os
from os.path import isfile, join
import poplib
import re
import requests_oauthlib
import socket
import ssl
import sys
@ -43,7 +45,6 @@ from typing import List, Tuple
# import User model, which may be a custom model
User = get_user_model()
STRIPPED_SUBJECT_STRINGS = [
"Re: ",
"Fw: ",
@ -225,6 +226,99 @@ def imap_sync(q, logger, server):
server.logout()
def imap_oauth_sync(q, logger, server):
"""
IMAP eMail server with OAUTH authentication.
Only tested against O365 implementation
Uses HELPDESK OAUTH Dict in Settings.
"""
try:
logger.debug("Start Mailbox polling via IMAP OAUTH")
client = oauth2lib.BackendApplicationClient(
client_id=settings.HELPDESK_OAUTH["client_id"],
scope=settings.HELPDESK_OAUTH["scope"],
)
oauth = requests_oauthlib.OAuth2Session(client=client)
token = oauth.fetch_token(
token_url=settings.HELPDESK_OAUTH["token_url"],
client_id=settings.HELPDESK_OAUTH["client_id"],
client_secret=settings.HELPDESK_OAUTH["secret"],
include_client_id=True,
)
server.debug = settings.HELPDESK_IMAP_DEBUG_LEVEL
# TODO: Perhaps store the authentication string template externally? Settings? Queue Table?
server.authenticate(
"XOAUTH2",
lambda x: f"user={q.email_box_user}\x01auth=Bearer {token['access_token']}\x01\x01".encode(),
)
# Select the Inbound Mailbox folder
server.select(q.email_box_imap_folder)
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 as e2:
logger.error(
f"IMAP login failed due to SSL error. (This is often due to a timeout): {e2}", exc_info=True
)
server.logout()
sys.exit()
try:
data = server.search(None, 'NOT', 'DELETED')[1]
if data:
msgnums = data[0].split()
logger.info(f"Found {len(msgnums)} message(s) on IMAP server")
for num in msgnums:
logger.info(f"Processing message {num}")
data = server.fetch(num, '(RFC822)')[1]
full_message = encoding.force_str(data[0][1], errors='replace')
try:
ticket = object_from_message(message=full_message, queue=q, logger=logger)
except IgnoreTicketException as itex:
logger.warn(f"Message {num} was ignored. {itex}")
except DeleteIgnoredTicketException:
server.store(num, '+FLAGS', '\\Deleted')
logger.warn("Message %s was ignored and deleted from IMAP server" % num)
except TypeError as te:
# Log the error with stacktrace to help identify what went wrong
logger.error(f"Unexpected error processing message: {te}", exc_info=True)
else:
if ticket:
server.store(num, '+FLAGS', '\\Deleted')
logger.info(
"Successfully processed message %s, deleted from IMAP server" % num)
else:
logger.warn(
"Message %s was not successfully processed, and will be left on IMAP server" % num)
except imaplib.IMAP4.error:
logger.error(
"IMAP retrieve failed. Is the folder '%s' spelled correctly, and does it exist on the server?",
q.email_box_imap_folder
)
# Purged Flagged Messages & Logout
server.expunge()
server.close()
server.logout()
def process_queue(q, logger):
logger.info("***** %s: Begin processing mail for django-helpdesk" % ctime())
@ -272,7 +366,18 @@ def process_queue(q, logger):
'init': imaplib.IMAP4,
},
'sync': imap_sync
}
},
'oauth': {
'ssl': {
'port': 993,
'init': imaplib.IMAP4_SSL,
},
'insecure': {
'port': 143,
'init': imaplib.IMAP4,
},
'sync': imap_oauth_sync
},
}
if email_box_type in mail_defaults:
encryption = 'insecure'
@ -361,7 +466,6 @@ def is_autoreply(message):
def create_ticket_cc(ticket, cc_list):
if not cc_list:
return []
@ -393,7 +497,6 @@ def create_ticket_cc(ticket, cc_list):
def create_object_from_email_message(message, ticket_id, payload, files, logger):
ticket, previous_followup, new = None, None, False
now = timezone.now()
@ -538,9 +641,9 @@ def send_info_email(message_id: str, f: FollowUp, ticket: Ticket, context: dict,
def get_ticket_id_from_subject_slug(
queue_slug: str,
subject: str,
logger: logging.Logger
queue_slug: str,
subject: str,
logger: logging.Logger
) -> typing.Optional[int]:
"""Get a ticket id from the subject string
@ -559,8 +662,8 @@ def get_ticket_id_from_subject_slug(
def add_file_if_always_save_incoming_email_message(
files_,
message: str
files_,
message: str
) -> None:
"""When `settings.HELPDESK_ALWAYS_SAVE_INCOMING_EMAIL_MESSAGE` is `True`
add a file to the files_ list"""

18
helpdesk/migrations/0037_alter_queue_email_box_type.py Normal file
View File

@ -0,0 +1,18 @@
# Generated by Django 3.2.16 on 2023-03-25 15:45
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('helpdesk', '0036_add_attachment_validator'),
]
operations = [
migrations.AlterField(
model_name='queue',
name='email_box_type',
field=models.CharField(blank=True, choices=[('pop3', 'POP 3'), ('imap', 'IMAP'), ('oauth', 'IMAP OAUTH'), ('local', 'Local Directory')], help_text='E-Mail server type for creating tickets automatically from a mailbox - both POP3 and IMAP are supported, as well as reading from a local directory.', max_length=5, null=True, verbose_name='E-Mail Box Type'),
),
]

4
helpdesk/models.py
View File

@ -175,7 +175,9 @@ class Queue(models.Model):
email_box_type = models.CharField(
_('E-Mail Box Type'),
max_length=5,
choices=(('pop3', _('POP 3')), ('imap', _('IMAP')),
choices=(('pop3', _('POP 3')),
('imap', _('IMAP')),
('oauth', _('IMAP OAUTH')),
('local', _('Local Directory'))),
blank=True,
null=True,

16
helpdesk/settings.py
View File

@ -249,3 +249,19 @@ HELPDESK_FULL_FIRST_MESSAGE_FROM_EMAIL = getattr(
# (which gets stripped/corrupted otherwise)
HELPDESK_ALWAYS_SAVE_INCOMING_EMAIL_MESSAGE = getattr(
settings, "HELPDESK_ALWAYS_SAVE_INCOMING_EMAIL_MESSAGE", False)
#######################
# email OAUTH #
#######################
HELPDESK_OAUTH = getattr(
settings, 'HELPDESK_OAUTH', {
"token_url": "",
"client_id": "",
"secret": "",
"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)

218
helpdesk/tests/test_get_email.py
View File

@ -1,6 +1,5 @@
# -*- coding: utf-8 -*-
from django.contrib.auth.hashers import make_password
from django.contrib.auth.models import User
from django.core import mail
@ -16,29 +15,31 @@ from helpdesk.models import Attachment, FollowUp, FollowUpAttachment, IgnoreEmai
from helpdesk.tests import utils
import itertools
import logging
from oauthlib.oauth2 import BackendApplicationClient
import os
from shutil import rmtree
import sys
from tempfile import mkdtemp
import time
import typing
from unittest import mock
THIS_DIR = os.path.dirname(os.path.abspath(__file__))
# class A addresses can't have first octet of 0
unrouted_socks_server = "0.0.0.1"
unrouted_email_server = "0.0.0.1"
# the last user port, reserved by IANA
unused_port = "49151"
fake_time = time.time()
class GetEmailCommonTests(TestCase):
def setUp(self):
self.queue_public = Queue.objects.create(title='Test', slug='test')
self.logger = logging.getLogger('helpdesk')
# tests correct syntax for command line option
def test_get_email_quiet_option(self):
"""Test quiet option is properly propagated"""
@ -60,7 +61,6 @@ class GetEmailCommonTests(TestCase):
test_email = fd.read()
ticket = helpdesk.email.object_from_message(
test_email, self.queue_public, self.logger)
# title got truncated because of max_lengh of the model.title field
assert ticket.title == (
"Attachment without body - and a loooooooooooooooooooooooooooooooooo"
@ -283,6 +283,17 @@ class GetEmailParametricTemplate(object):
self.queue_public = Queue.objects.create(**kwargs)
self.token = {
'token_type': 'Bearer',
'access_token': 'asdfoiw37850234lkjsdfsdf',
'refresh_token': 'sldvafkjw34509s8dfsdf',
'expires_in': '3600',
'expires_at': fake_time + 3600,
}
self.client_id = 'foo'
self.client = BackendApplicationClient(self.client_id)
def tearDown(self):
rmtree(self.temp_logdir)
@ -291,7 +302,6 @@ class GetEmailParametricTemplate(object):
"""Tests reading plain text 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 Django docs:
# https://docs.djangoproject.com/en/1.10/ref/unicode/
test_email_from = "Arnbjörg Ráðormsdóttir <arnbjorg@example.com>"
@ -355,7 +365,6 @@ class GetEmailParametricTemplate(object):
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(
@ -365,6 +374,43 @@ class GetEmailParametricTemplate(object):
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)
@ -452,6 +498,44 @@ class GetEmailParametricTemplate(object):
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])
@ -543,6 +627,44 @@ class GetEmailParametricTemplate(object):
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)
@ -651,6 +773,7 @@ class GetEmailParametricTemplate(object):
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
@ -672,6 +795,44 @@ class GetEmailParametricTemplate(object):
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)
@ -751,6 +912,7 @@ class GetEmailParametricTemplate(object):
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
@ -771,6 +933,43 @@ class GetEmailParametricTemplate(object):
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(
@ -876,7 +1075,6 @@ class GetEmailCCHandling(TestCase):
def test_read_email_cc(self):
"""Tests reading plain text emails from a queue and adding to a ticket,
particularly to test appropriate handling of CC'd emails."""
# first, check that test ticket exists
ticket1 = get_object_or_404(Ticket, pk=1)
self.assertEqual(ticket1.ticket_for_url, "CC-1")
@ -887,7 +1085,6 @@ class GetEmailCCHandling(TestCase):
self.assertEqual(ccstaff.user, User.objects.get(username='staff'))
self.assertEqual(ticket1.assigned_to,
User.objects.get(username='assigned'))
# example email text from Django docs:
# https://docs.djangoproject.com/en/1.10/ref/unicode/
test_email_from = "submitter@example.com"
@ -918,7 +1115,6 @@ class GetEmailCCHandling(TestCase):
mocked_listdir.assert_called_with('/var/lib/mail/helpdesk/')
mocked_isfile.assert_any_call('/var/lib/mail/helpdesk/filename1')
# 9 unique email addresses are CC'd when all is done
self.assertEqual(len(TicketCC.objects.filter(ticket=1)), 9)
# next we make sure no duplicates were added, and the
@ -942,16 +1138,12 @@ class GetEmailCCHandling(TestCase):
cc9 = get_object_or_404(TicketCC, pk=9)
self.assertEqual(cc9.user, User.objects.get(username='observer'))
self.assertEqual(cc9.email, "observer@example.com")
# build matrix of test cases
case_methods = [c[0] for c in Queue._meta.get_field('email_box_type').choices]
# uncomment if you want to run tests with socks - which is much slover
# case_socks = [False] + [c[0] for c in Queue._meta.get_field('socks_proxy_type').choices]
case_socks = [False]
case_matrix = list(itertools.product(case_methods, case_socks))
# Populate TestCases from the matrix of parameters
thismodule = sys.modules[__name__]
for method, socks in case_matrix:

4
quicktest.py
View File

@ -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

1
requirements-testing.txt
View File

@ -1,6 +1,5 @@
pysocks
pycodestyle
codecov
coverage
argparse
pbr

2
requirements.txt
View File

@ -12,3 +12,5 @@ pinax_teams
djangorestframework
django-model-utils
django-cleanup
oauthlib
requests_oauthlib