altf4arnold/datatracker
1
0
Fork 0
Code Issues Pull requests Actions 4 Packages Projects Releases Wiki Activity

Accept/replace invalid Unicode bytes when processing ipr response emails. Fixes #3489. Commit ready for merge.

Browse source 
- Legacy-Id: 19766
This commit is contained in:
Jennifer Richards 2021-12-09 19:29:20 +00:00
parent fd0df6f619
commit 310ea57bc5
4 changed files with 64 additions and 34 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
ietf/ipr/mail.py
View file

@ -11,7 +11,7 @@ import pytz
import re
from django.template.loader import render_to_string
from django.utils.encoding import force_text, force_str
from django.utils.encoding import force_text, force_bytes
import debug # pyflakes:ignore
@ -174,7 +174,7 @@ def process_response_email(msg):
a matching value in the reply_to field, associated to an IPR disclosure through
IprEvent. Create a Message object for the incoming message and associate it to
the original message via new IprEvent"""
message = email.message_from_string(force_str(msg))
message = email.message_from_bytes(force_bytes(msg))
to = message.get('To', '')
# exit if this isn't a response we're interested in (with plus addressing)

4
ietf/ipr/management/commands/process_email.py
View file

@ -25,10 +25,8 @@ class Command(EmailOnFailureCommand):
email = options.get('email', None)
binary_input = io.open(email, 'rb') if email else sys.stdin.buffer
self.msg_bytes = binary_input.read()
msg = self.msg_bytes.decode()
try:
process_response_email(msg)
process_response_email(self.msg_bytes)
except ValueError as e:
raise CommandError(e)

25
ietf/ipr/management/tests.py
View file

@ -18,9 +18,10 @@ class ProcessEmailTests(TestCase):
with name_of_file_containing('contents') as filename:
call_command('process_email', email_file=filename)
self.assertEqual(process_mock.call_count, 1, 'process_response_email should be called once')
(msg,) = process_mock.call_args.args
self.assertEqual(
process_mock.call_args.args,
('contents',),
msg.decode(),
'contents',
'process_response_email should receive the correct contents'
)
@ -52,16 +53,15 @@ class ProcessEmailTests(TestCase):
@mock.patch('ietf.utils.management.base.send_smtp')
@mock.patch('ietf.ipr.management.commands.process_email.process_response_email')
def test_invalid_character_encodings(self, process_mock, send_smtp_mock):
"""The process_email command should attach messages with invalid encoding when using a file input"""
"""The process_email command should accept messages with invalid encoding when using a file input"""
invalid_characters = b'\xfe\xff'
with name_of_file_containing(invalid_characters, mode='wb') as filename:
call_command('process_email', email_file=filename)
self.assertFalse(process_mock.called) # should not even try to process illegally encoded messages
self.assertTrue(send_smtp_mock.called)
(msg,) = send_smtp_mock.call_args.args
parts = msg.get_payload()
self.assertEqual(len(parts), 3, 'Error email should contain message, traceback, and original message')
self.assertFalse(send_smtp_mock.called) # should not send an error email
self.assertTrue(process_mock.called)
(msg,) = process_mock.call_args.args
self.assertEqual(msg, invalid_characters, 'Invalid unicode should be passed to process_email()')
@mock.patch.object(sys.stdin.buffer, 'read')
@mock.patch('ietf.utils.management.base.send_smtp')
@ -72,8 +72,7 @@ class ProcessEmailTests(TestCase):
stdin_read_mock.return_value = invalid_characters
call_command('process_email')
self.assertFalse(process_mock.called) # should not even try to process illegally encoded messages
self.assertTrue(send_smtp_mock.called)
(msg,) = send_smtp_mock.call_args.args
parts = msg.get_payload()
self.assertEqual(len(parts), 3, 'Error email should contain message, traceback, and original message')
self.assertFalse(send_smtp_mock.called) # should not send an error email
self.assertTrue(process_mock.called)
(msg,) = process_mock.call_args.args
self.assertEqual(msg, invalid_characters, 'Invalid unicode should be passed to process_email()')

65
ietf/ipr/tests.py
View file

@ -592,8 +592,7 @@ I would like to revoke this declaration.
self.assertEqual(len(outbox),2)
self.assertIn('Secretariat on '+ipr.get_latest_event_submitted().time.strftime("%Y-%m-%d"), get_payload_text(outbox[1]).replace('\n',' '))
def test_process_response_email(self):
# first send a mail
def send_ipr_email_helper(self):
ipr = HolderIprDisclosureFactory()
url = urlreverse('ietf.ipr.views.email',kwargs={ "id": ipr.id })
self.client.login(username="secretary", password="secretary+password")
@ -607,7 +606,6 @@ I would like to revoke this declaration.
response_due=yesterday.isoformat())
empty_outbox()
r = self.client.post(url,data,follow=True)
#print r.content
self.assertEqual(r.status_code,200)
q = Message.objects.filter(reply_to=data['reply_to'])
self.assertEqual(q.count(),1)
@ -615,19 +613,25 @@ I would like to revoke this declaration.
self.assertTrue(event.response_past_due())
self.assertEqual(len(outbox), 1)
self.assertTrue('joe@test.com' in outbox[0]['To'])
return data['reply_to'], event
uninteresting_ipr_message_strings = [
("To: {to}\nCc: {cc}\nFrom: joe@test.com\nDate: {date}\nSubject: test\n"),
("Cc: {cc}\nFrom: joe@test.com\nDate: {date}\nSubject: test\n"), # no To
("To: {to}\nFrom: joe@test.com\nDate: {date}\nSubject: test\n"), # no Cc
("From: joe@test.com\nDate: {date}\nSubject: test\n"), # no To or Cc
("Cc: {cc}\nDate: {date}\nSubject: test\n"), # no To
("To: {to}\nDate: {date}\nSubject: test\n"), # no Cc
("Date: {date}\nSubject: test\n"), # no To or Cc
]
def test_process_response_email(self):
# first send a mail
reply_to, event = self.send_ipr_email_helper()
# test process response uninteresting messages
addrs = gather_address_lists('ipr_disclosure_submitted').as_strings()
uninteresting_message_strings = [
("To: {to}\nCc: {cc}\nFrom: joe@test.com\nDate: {date}\nSubject: test\n"),
("Cc: {cc}\nFrom: joe@test.com\nDate: {date}\nSubject: test\n"), # no To
("To: {to}\nFrom: joe@test.com\nDate: {date}\nSubject: test\n"), # no Cc
("From: joe@test.com\nDate: {date}\nSubject: test\n"), # no To or Cc
("Cc: {cc}\nDate: {date}\nSubject: test\n"), # no To
("To: {to}\nDate: {date}\nSubject: test\n"), # no Cc
("Date: {date}\nSubject: test\n"), # no To or Cc
]
for message_string in uninteresting_message_strings:
for message_string in self.uninteresting_ipr_message_strings:
result = process_response_email(
message_string.format(
to=addrs.to,
@ -642,12 +646,41 @@ I would like to revoke this declaration.
From: joe@test.com
Date: {}
Subject: test
""".format(data['reply_to'],datetime.datetime.now().ctime())
""".format(reply_to, datetime.datetime.now().ctime())
result = process_response_email(message_string)
self.assertIsInstance(result,Message)
self.assertIsInstance(result, Message)
self.assertFalse(event.response_past_due())
def test_process_response_email_with_invalid_encoding(self):
"""Interesting emails with invalid encoding should be handled"""
reply_to, _ = self.send_ipr_email_helper()
# test process response
message_string = """To: {}
From: joe@test.com
Date: {}
Subject: test
""".format(reply_to, datetime.datetime.now().ctime())
message_bytes = message_string.encode('utf8') + b'\nInvalid stuff: \xfe\xff\n'
result = process_response_email(message_bytes)
self.assertIsInstance(result, Message)
# \ufffd is a rhombus character with an inverse ?, used to replace invalid characters
self.assertEqual(result.body, 'Invalid stuff: \ufffd\ufffd\n\n', # not sure where the extra \n is from
'Invalid characters should be replaced with \ufffd characters')
def test_process_response_email_uninteresting_with_invalid_encoding(self):
"""Uninteresting emails with invalid encoding should be quietly dropped"""
self.send_ipr_email_helper()
addrs = gather_address_lists('ipr_disclosure_submitted').as_strings()
for message_string in self.uninteresting_ipr_message_strings:
message_bytes = message_string.format(
to=addrs.to,
cc=addrs.cc,
date=datetime.datetime.now().ctime(),
).encode('utf8') + b'\nInvalid stuff: \xfe\xff\n'
result = process_response_email(message_bytes)
self.assertIsNone(result)
def test_ajax_search(self):
url = urlreverse('ietf.ipr.views.ajax_search')
response=self.client.get(url+'?q=disclosure')