Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 16 additions & 5 deletions ietf/ipr/mail.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@
import re

from email import message_from_bytes
from email.utils import parsedate_tz
from email.utils import parsedate_tz, parseaddr

from django.template.loader import render_to_string
from django.utils.encoding import force_str, force_bytes
Expand Down Expand Up @@ -176,6 +176,17 @@ class UndeliverableIprResponseError(Exception):
"""Response email could not be delivered and should be treated as an error"""


def extract_valid_response_email_to_address(to: str) -> str | None:
# exit if this isn't a response we're interested in (with plus addressing)
local, domain = get_base_ipr_request_address().split("@")
valid_pattern = fr"{local}\+[a-zA-Z0-9_\-]{{16}}@{domain}$"
to_name, to_addr = parseaddr(to)
if (to_name, to_addr) == ("", ""):
return None # entirely invalid address
is_valid = re.fullmatch(valid_pattern, to_addr) is not None
return to_addr if is_valid else None


def process_response_email(msg):

This comment was marked as spam.

"""Save an incoming IPR response email message

Expand All @@ -189,13 +200,13 @@ def process_response_email(msg):
the original message via new IprEvent
"""
message = message_from_bytes(force_bytes(msg))
to = message.get('To', '')
raw_to = message.get('To', '')

# exit if this isn't a response we're interested in (with plus addressing)
local, domain = get_base_ipr_request_address().split('@')
if not re.match(r'^{}\+[a-zA-Z0-9_\-]{}@{}'.format(local,'{16}',domain),to):
to = extract_valid_response_email_to_address(raw_to)
if to is None:
_from = message.get("From", "<unknown>")
log(f"Ignoring IPR email without a message identifier from {_from} to {to}")
log(f"Ignoring IPR email without a message identifier from {_from} to {raw_to}")
return

try:
Expand Down
167 changes: 17 additions & 150 deletions ietf/ipr/tests.py
Original file line number Diff line number Diff line change
@@ -1,55 +1,52 @@
# Copyright The IETF Trust 2009-2023, All Rights Reserved
# -*- coding: utf-8 -*-

# Copyright The IETF Trust 2009-2026, All Rights Reserved

import datetime
import json
from unittest import mock
import re

from pyquery import PyQuery
from urllib.parse import quote, urlparse
from zoneinfo import ZoneInfo

from django.conf import settings
from django.db.models import Max
from django.test.utils import override_settings
from django.urls import reverse as urlreverse
from django.utils import timezone
from pyquery import PyQuery

from django.db.models import Max

import debug # pyflakes:ignore

import debug # pyflakes:ignore
from ietf.api.views import EmailIngestionError
from ietf.doc.factories import (
DocumentFactory,
NewRevisionDocEventFactory,
RfcFactory,
WgDraftFactory,
WgRfcFactory,
RfcFactory,
NewRevisionDocEventFactory
)
from ietf.doc.utils import prettify_std_name
from ietf.group.factories import RoleFactory
from ietf.ipr.factories import (
HolderIprDisclosureFactory,
GenericIprDisclosureFactory,
HolderIprDisclosureFactory,
IprDisclosureBaseFactory,
IprDocRelFactory,
IprEventFactory,
ThirdPartyIprDisclosureFactory
ThirdPartyIprDisclosureFactory,
)
from ietf.ipr.forms import DraftForm, HolderIprDisclosureForm
from ietf.ipr.mail import (process_response_email, get_reply_to, get_update_submitter_emails,
get_pseudo_submitter, get_holders, get_update_cc_addrs, UndeliverableIprResponseError)
from ietf.ipr.models import (IprDisclosureBase, GenericIprDisclosure, HolderIprDisclosure, RemovedIprDisclosure,
ThirdPartyIprDisclosure, IprEvent)
from ietf.ipr.mail import UndeliverableIprResponseError
from ietf.ipr.models import (
GenericIprDisclosure,
HolderIprDisclosure,
IprDisclosureBase,
RemovedIprDisclosure,
ThirdPartyIprDisclosure,
)
from ietf.ipr.templatetags.ipr_filters import no_revisions_message
from ietf.ipr.utils import get_genitive, get_ipr_summary, ingest_response_email
from ietf.mailtrigger.utils import gather_address_lists
from ietf.message.factories import MessageFactory
from ietf.message.models import Message
from ietf.person.factories import PersonFactory
from ietf.utils.mail import outbox, empty_outbox, get_payload_text
from ietf.utils.mail import empty_outbox, get_payload_text, outbox
from ietf.utils.test_utils import TestCase, login_testing_unauthorized
from ietf.utils.text import text_to_dict
from ietf.utils.timezone import date_today
Expand All @@ -75,39 +72,10 @@ def test_get_genitive(self):
self.assertEqual(get_genitive("Cisco"),"Cisco's")
self.assertEqual(get_genitive("Ross"),"Ross'")

def test_get_holders(self):
ipr = HolderIprDisclosureFactory()
update = HolderIprDisclosureFactory(updates=[ipr,])
result = get_holders(update)
self.assertEqual(set(result),set([ipr.holder_contact_email,update.holder_contact_email]))

def test_get_ipr_summary(self):
ipr = HolderIprDisclosureFactory(docs=[WgDraftFactory(),])
self.assertEqual(get_ipr_summary(ipr),ipr.docs.first().name)

def test_get_pseudo_submitter(self):
ipr = HolderIprDisclosureFactory()
self.assertEqual(get_pseudo_submitter(ipr),(ipr.submitter_name,ipr.submitter_email))
ipr.submitter_name=''
ipr.submitter_email=''
self.assertEqual(get_pseudo_submitter(ipr),(ipr.holder_contact_name,ipr.holder_contact_email))
ipr.holder_contact_name=''
ipr.holder_contact_email=''
self.assertEqual(get_pseudo_submitter(ipr),('UNKNOWN NAME - NEED ASSISTANCE HERE','UNKNOWN EMAIL - NEED ASSISTANCE HERE'))

def test_get_update_cc_addrs(self):
ipr = HolderIprDisclosureFactory()
update = HolderIprDisclosureFactory(updates=[ipr,])
result = get_update_cc_addrs(update)
self.assertEqual(set(result.split(',')),set([update.holder_contact_email,ipr.submitter_email,ipr.holder_contact_email]))

def test_get_update_submitter_emails(self):
ipr = HolderIprDisclosureFactory()
update = HolderIprDisclosureFactory(updates=[ipr,])
messages = get_update_submitter_emails(update)
self.assertEqual(len(messages),1)
self.assertTrue(messages[0].startswith('To: %s' % ipr.submitter_email))

def test_showlist(self):
for disc_factory_type in (HolderIprDisclosureFactory, GenericIprDisclosureFactory, ThirdPartyIprDisclosureFactory):
ipr = disc_factory_type(state_id="removed")
Expand Down Expand Up @@ -749,107 +717,6 @@ def test_notify_generic(self):
)
self.assertIn(f'{settings.IDTRACKER_BASE_URL}{urlreverse("ietf.ipr.views.showlist")}', get_payload_text(outbox[1]).replace('\n',' '))

def send_ipr_email_helper(self) -> tuple[str, IprEvent, HolderIprDisclosure]:
ipr = HolderIprDisclosureFactory.create() # call create() explicitly so mypy sees correct type
url = urlreverse('ietf.ipr.views.email',kwargs={ "id": ipr.id })
self.client.login(username="secretary", password="secretary+password")
yesterday = date_today() - datetime.timedelta(1)
data = dict(
to='joe@test.com',
frm='ietf-ipr@ietf.org',
subject='test',
reply_to=get_reply_to(),
body='Testing.',
response_due=yesterday.isoformat())
empty_outbox()
r = self.client.post(url,data,follow=True)
self.assertEqual(r.status_code,200)
q = Message.objects.filter(reply_to=data['reply_to'])
self.assertEqual(q.count(),1)
event = q[0].msgevents.first()
assert event is not None
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, ipr

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()
for message_string in self.uninteresting_ipr_message_strings:
process_response_email(
message_string.format(
to=addrs.to,
cc=addrs.cc,
date=timezone.now().ctime()
)
)

# test process response
message_string = """To: {}
From: joe@test.com
Date: {}
Subject: test
""".format(reply_to, timezone.now().ctime())
process_response_email(message_string)
self.assertFalse(event.response_past_due())

# test with an unmatchable message identifier
bad_reply_to = re.sub(
r"\+.{16}@",
'+0123456789abcdef@',
reply_to,
)
self.assertNotEqual(reply_to, bad_reply_to)
message_string = f"""To: {bad_reply_to}
From: joe@test.com
Date: {timezone.now().ctime()}
Subject: test
"""
with self.assertRaises(UndeliverableIprResponseError):
process_response_email(message_string)

def test_process_response_email_with_invalid_encoding(self):
"""Interesting emails with invalid encoding should be handled"""
reply_to, _, disclosure = self.send_ipr_email_helper()
# test process response
message_string = """To: {}
From: joe@test.com
Date: {}
Subject: test
""".format(reply_to, timezone.now().ctime())
message_bytes = message_string.encode('utf8') + b'\nInvalid stuff: \xfe\xff\n'
process_response_email(message_bytes)
result = IprEvent.objects.filter(disclosure=disclosure).first().message # newest
# \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=timezone.now().ctime(),
).encode('utf8') + b'\nInvalid stuff: \xfe\xff\n'
process_response_email(message_bytes)

@override_settings(ADMINS=(("Some Admin", "admin@example.com"),))
@mock.patch("ietf.ipr.utils.process_response_email")
def test_ingest_response_email(self, mock_process_response_email):
Expand Down
Loading