You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@allura.apache.org by cc...@apache.org on 2026/08/20 19:51:29 UTC
(allura) branch cc/6054-custom-header created (now 0ca0fbef2)
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a change to branch cc/6054-custom-header
in repository https://gitbox.apache.org/repos/asf/allura.git
at 0ca0fbef2 [#6054] Simplify trusted MX header verification
This branch includes the following new commits:
new ef91bfbfb [#6054] Verify inbound senders with trusted MX header
new 0ca0fbef2 [#6054] Simplify trusted MX header verification
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(allura) 01/02: [#6054] Verify inbound senders with trusted MX header
Posted by cc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch cc/6054-custom-header
in repository https://gitbox.apache.org/repos/asf/allura.git
commit ef91bfbfbedede247bcfde2558298f66841f9097
Author: Carlos Cruz <ca...@slashdotmedia.com>
AuthorDate: Fri Aug 14 17:01:18 2026 -0600
[#6054] Verify inbound senders with trusted MX header
---
Allura/allura/lib/mail_util.py | 233 +++++++++++++++++++++++++++-
Allura/allura/tests/test_mail_util.py | 276 ++++++++++++++++++++++++++++++++++
Allura/allura/tests/test_tasks.py | 58 +++++++
Allura/development.ini | 5 +
4 files changed, 571 insertions(+), 1 deletion(-)
diff --git a/Allura/allura/lib/mail_util.py b/Allura/allura/lib/mail_util.py
index d086dc9f4..06ee6f664 100644
--- a/Allura/allura/lib/mail_util.py
+++ b/Allura/allura/lib/mail_util.py
@@ -24,6 +24,7 @@
from email import header
from email.message import EmailMessage
+import idna
import six
import tg
from paste.deploy.converters import asbool, asint, aslist
@@ -42,11 +43,23 @@
common_suffix='forgemail.domain',
common_suffix_alt='forgemail.domain.alternates',
return_path='forgemail.return_path',
+ sender_authentication_enabled='forgemail.sender_authentication.enabled',
+ sender_authentication_header='forgemail.sender_authentication.header',
)
EMAIL_VALIDATOR = fev.Email(not_empty=True)
# http://www.jebriggs.com/blog/2010/07/smtp-maximum-line-lengths/
MAX_MAIL_LINE_OCTETS = 990
+MAX_SENDER_AUTHENTICATION_HEADER_OCTETS = 1024
+MAX_FROM_HEADER_OCTETS = 2 * 1024
+SENDER_AUTHENTICATION_HEADER = 'X-SourceForge-Sender-Auth'
+RE_HEADER_NAME = re.compile(r'(?a)\A[A-Za-z0-9][A-Za-z0-9-]{0,126}\Z')
+RE_DMARC_SENDER_AUTHENTICATION = re.compile(
+ r'(?a)\Av=1;[ \t]+dmarc=pass;[ \t]+'
+ r'from-domain=([A-Za-z0-9.-]+)\Z')
+RE_SPF_SENDER_AUTHENTICATION = re.compile(
+ r'(?a)\Av=1;[ \t]+result=pass;[ \t]+method=spf-aligned;[ \t]+'
+ r'from-domain=([A-Za-z0-9.-]+)\Z')
email_policy = email.policy.SMTP + email.policy.strict
@@ -144,6 +157,20 @@ def parse_message(data):
# Extract relevant data
result = {}
result['multipart'] = multipart = msg.is_multipart()
+ # dict(msg) intentionally retains the longstanding, first-header-wins view
+ # used by message handlers. Sender authentication separately preserves all
+ # From and configured authentication fields so duplicates fail closed.
+ raw_headers = list(msg.raw_items())
+ result['from_headers'] = [
+ value for name, value in raw_headers if name.casefold() == 'from']
+ authentication_header = config.get('sender_authentication_header', '')
+ if isinstance(authentication_header, str):
+ authentication_header = authentication_header.strip().casefold()
+ else:
+ authentication_header = ''
+ result['sender_authentication_headers'] = [
+ value for name, value in raw_headers
+ if authentication_header and name.casefold() == authentication_header]
result['headers'] = dict(msg)
result['message_id'] = _parse_message_id(msg.get('Message-ID'))
result['in_reply_to'] = _parse_message_id(msg.get('In-Reply-To'))
@@ -176,7 +203,179 @@ def parse_message(data):
return result
-def identify_sender(peer, email_address, headers, msg):
+class SenderAuthenticationError(ValueError):
+
+ def __init__(self, reason):
+ self.reason = reason
+ super().__init__(reason)
+
+
+def _unfold_sender_authentication_header(value, max_octets):
+ if not isinstance(value, str):
+ raise SenderAuthenticationError('header_type')
+ try:
+ if len(value.encode('utf-8')) > max_octets:
+ raise SenderAuthenticationError('header_too_long')
+ except UnicodeEncodeError:
+ raise SenderAuthenticationError('header_unicode')
+
+ unfolded = []
+ i = 0
+ while i < len(value):
+ char = value[i]
+ if char in '\r\n':
+ if char == '\r':
+ if i + 1 >= len(value) or value[i + 1] != '\n':
+ raise SenderAuthenticationError('header_bad_fold')
+ i += 2
+ else:
+ i += 1
+ if i >= len(value) or value[i] not in ' \t':
+ raise SenderAuthenticationError('header_bad_fold')
+ while i < len(value) and value[i] in ' \t':
+ i += 1
+ unfolded.append(' ')
+ continue
+ codepoint = ord(char)
+ if ((codepoint < 0x20 and char != '\t') or codepoint == 0x7f or
+ 0x80 <= codepoint <= 0x9f):
+ raise SenderAuthenticationError('header_control')
+ unfolded.append(char)
+ i += 1
+
+ value = ''.join(unfolded).strip(' \t')
+ if not value:
+ raise SenderAuthenticationError('header_empty')
+ return value
+
+
+def _normalize_sender_domain(domain):
+ if (not isinstance(domain, str) or not domain or
+ domain.startswith('.') or domain.endswith('.')):
+ raise SenderAuthenticationError('domain_invalid')
+ try:
+ normalized = idna.encode(
+ domain,
+ uts46=True,
+ transitional=False,
+ std3_rules=True,
+ ).decode('ascii').lower()
+ except (idna.IDNAError, UnicodeError):
+ raise SenderAuthenticationError('domain_invalid')
+ labels = normalized.split('.')
+ if (len(normalized) > 253 or '..' in normalized or
+ any(not label or len(label) > 63 for label in labels)):
+ raise SenderAuthenticationError('domain_invalid')
+ return normalized
+
+
+def _parse_sender_authentication_header(value, from_domain):
+ value = _unfold_sender_authentication_header(
+ value, MAX_SENDER_AUTHENTICATION_HEADER_OCTETS)
+ for method, pattern in (
+ ('dmarc', RE_DMARC_SENDER_AUTHENTICATION),
+ ('spf_aligned', RE_SPF_SENDER_AUTHENTICATION)):
+ match = pattern.fullmatch(value)
+ if match:
+ break
+ else:
+ raise SenderAuthenticationError('sender_authentication_syntax')
+
+ if _normalize_sender_domain(match.group(1)) != from_domain:
+ raise SenderAuthenticationError('from_domain_mismatch')
+ return method
+
+
+def _parse_from_mailbox(value):
+ value = _unfold_sender_authentication_header(
+ value, MAX_FROM_HEADER_OCTETS)
+ parsed = email.parser.HeaderParser(policy=email.policy.default).parsestr(
+ f'From: {value}\n\n')
+ field = parsed['From']
+ if field is None or field.defects:
+ raise SenderAuthenticationError('from_defect')
+ addresses = tuple(field.addresses)
+ groups = tuple(field.groups)
+ if (len(addresses) != 1 or
+ any(group.display_name is not None for group in groups)):
+ raise SenderAuthenticationError('from_mailbox_count')
+ address = addresses[0]
+ if not address.username or not address.domain:
+ raise SenderAuthenticationError('from_invalid')
+ mailbox = address.addr_spec
+ if len(mailbox.encode('utf-8')) > 320:
+ raise SenderAuthenticationError('from_too_long')
+ return mailbox, _normalize_sender_domain(address.domain)
+
+
+def _configured_sender_authentication_header():
+ header_name = config.get('sender_authentication_header', '')
+ if not isinstance(header_name, str):
+ raise SenderAuthenticationError('sender_authentication_header_config')
+ header_name = header_name.strip()
+ # Exim strips this reserved name before adding its own attestation. Treat
+ # the setting as a deployment-parity check; accepting an arbitrary name
+ # would let an inbound copy under that other name bypass Exim's stripping.
+ if (not RE_HEADER_NAME.fullmatch(header_name) or
+ header_name.casefold() !=
+ SENDER_AUTHENTICATION_HEADER.casefold()):
+ raise SenderAuthenticationError('sender_authentication_header_config')
+ return header_name
+
+
+def _sender_authentication_enabled():
+ value = config.get('sender_authentication_enabled', False)
+ if isinstance(value, bool):
+ return value
+ if isinstance(value, str):
+ value = value.strip().lower()
+ if value in {'true', 'yes', 'on', '1'}:
+ return True
+ if value in {'false', 'no', 'off', '0'}:
+ return False
+ raise ValueError('invalid sender authentication enabled value')
+
+
+def _authenticated_sender(msg):
+ from allura import model as M
+
+ anonymous = M.User.anonymous()
+ try:
+ _configured_sender_authentication_header()
+ if not isinstance(msg, dict):
+ raise SenderAuthenticationError('message_missing')
+ authentication_headers = msg.get('sender_authentication_headers')
+ from_headers = msg.get('from_headers')
+ if (not isinstance(authentication_headers, (list, tuple)) or
+ len(authentication_headers) != 1):
+ raise SenderAuthenticationError('sender_authentication_header_count')
+ if (not isinstance(from_headers, (list, tuple)) or
+ len(from_headers) != 1):
+ raise SenderAuthenticationError('from_count')
+
+ mailbox, from_domain = _parse_from_mailbox(from_headers[0])
+ authentication_method = _parse_sender_authentication_header(
+ authentication_headers[0], from_domain)
+
+ # DMARC or aligned SPF proves control of the From domain, not a
+ # cryptographic per-mailbox identity. The exact confirmed address is
+ # the application-side mapping for that domain-authenticated mailbox.
+ owners = {}
+ for address in M.EmailAddress.find({
+ 'email': mailbox,
+ 'confirmed': True,
+ }):
+ owner = address.claimed_by_user()
+ if owner is not None:
+ owners[owner._id] = owner
+ if len(owners) != 1:
+ raise SenderAuthenticationError('active_owner_count')
+ return next(iter(owners.values())), f'{authentication_method}_pass'
+ except SenderAuthenticationError as error:
+ return anonymous, error.reason
+
+
+def _legacy_identify_sender(email_address, headers):
from allura import model as M
# Dumb ID -- just look for email address claimed by a particular user
addr = M.EmailAddress.get(email=email_address, confirmed=True)
@@ -191,6 +390,38 @@ def identify_sender(peer, email_address, headers, msg):
return M.User.anonymous()
+def identify_sender(peer, email_address, headers, msg):
+ from allura import model as M
+
+ try:
+ enabled = _sender_authentication_enabled()
+ except ValueError:
+ log.error(
+ 'Invalid forgemail sender authentication enabled value; '
+ 'using anonymous sender')
+ return M.User.anonymous()
+
+ if not enabled:
+ return _legacy_identify_sender(email_address, headers)
+ try:
+ verified_user, reason = _authenticated_sender(msg)
+ except Exception:
+ log.exception('Unexpected inbound sender authentication error')
+ return M.User.anonymous()
+ peer_address = (
+ peer[0]
+ if isinstance(peer, (tuple, list)) and peer
+ else 'invalid')
+ log.info(
+ 'Inbound sender authentication enabled=true result=%s reason=%s '
+ 'peer=%s',
+ not verified_user.is_anonymous(),
+ reason,
+ peer_address,
+ )
+ return verified_user
+
+
def encode_email_part(content, content_type):
try:
# simplest email - plain ascii
diff --git a/Allura/allura/tests/test_mail_util.py b/Allura/allura/tests/test_mail_util.py
index be44647a2..05b4b1c59 100644
--- a/Allura/allura/tests/test_mail_util.py
+++ b/Allura/allura/tests/test_mail_util.py
@@ -24,6 +24,7 @@
from tg import config as tg_config
from smtplib import SMTP as SMTPClient
from alluratest.controller import setup_basic_test, setup_global_objects
+from allura import model as M
from allura.lib.utils import ConfigProxy
from allura.app import Application
from allura.lib.mail_util import (
@@ -212,6 +213,25 @@ def test_unicode_complex_message(self):
continue
assert isinstance(part['payload'], str), type(part['payload'])
+ @mock.patch.dict(tg_config, {
+ 'forgemail.sender_authentication.header': 'X-Test-Sender-Auth',
+ })
+ def test_parse_message_preserves_repeated_sender_authentication_headers(self):
+ msg = parse_message('''\
+From: first@example.com
+From: second@example.com
+X-Test-Sender-Auth: v=1; dmarc=pass; from-domain=example.com
+x-test-sender-auth:
+ v=1; result=pass; method=spf-aligned; from-domain=example.com
+
+body''')
+
+ assert msg['from_headers'] == ['first@example.com', 'second@example.com']
+ assert msg['sender_authentication_headers'] == [
+ 'v=1; dmarc=pass; from-domain=example.com',
+ '\n v=1; result=pass; method=spf-aligned; from-domain=example.com',
+ ]
+
class TestHeader:
@@ -321,6 +341,262 @@ def test_no_match(self, EA, User):
[mock.call(email='arg', confirmed=True), mock.call(email='from')])
+class TestAuthenticatedIdentifySender:
+
+ HEADER_NAME = 'X-SourceForge-Sender-Auth'
+ AUTH_CONFIG = {
+ 'forgemail.sender_authentication.enabled': 'true',
+ 'forgemail.sender_authentication.header': HEADER_NAME,
+ }
+ DMARC_PASS = 'v=1; dmarc=pass; from-domain=users.localhost'
+ SPF_ALIGNED_PASS = (
+ 'v=1; result=pass; method=spf-aligned; '
+ 'from-domain=users.localhost')
+
+ def setup_method(self, method):
+ setup_basic_test()
+ setup_global_objects()
+ ThreadLocalODMSession.flush_all()
+
+ def _message(self, from_headers=None, sender_authentication_headers=None,
+ header_name=None, extra_headers=None):
+ if from_headers is None:
+ from_headers = ['"Test Admin" <te...@users.localhost>']
+ if sender_authentication_headers is None:
+ sender_authentication_headers = [self.DMARC_PASS]
+ if header_name is None:
+ header_name = self.HEADER_NAME
+ lines = []
+ lines.extend(f'From: {value}' for value in from_headers)
+ lines.extend(
+ f'{header_name}: {value}'
+ for value in sender_authentication_headers)
+ lines.extend(extra_headers or [])
+ lines.extend(['Subject: inbound sender authentication test', '', 'body'])
+ return parse_message('\n'.join(lines))
+
+ def _identify(self, *, peer=('127.0.0.1', 2525),
+ mailfrom='test-admin@users.localhost', from_headers=None,
+ sender_authentication_headers=None, auth_config=None,
+ header_name=None, extra_headers=None):
+ if auth_config is None:
+ auth_config = self.AUTH_CONFIG
+ with mock.patch.dict(tg_config, auth_config):
+ msg = self._message(
+ from_headers, sender_authentication_headers, header_name,
+ extra_headers)
+ return identify_sender(peer, mailfrom, msg['headers'], msg)
+
+ def test_disabled_preserves_envelope_sender_lookup(self):
+ user = self._identify(
+ peer=('203.0.113.10', 2525),
+ sender_authentication_headers=[],
+ auth_config={
+ 'forgemail.sender_authentication.enabled': 'false',
+ 'forgemail.sender_authentication.header': self.HEADER_NAME,
+ })
+
+ assert user.username == 'test-admin'
+
+ @pytest.mark.parametrize('sender_authentication_header', [
+ DMARC_PASS,
+ SPF_ALIGNED_PASS,
+ 'v=1; dmarc=pass; from-domain=USERS.LOCALHOST',
+ ])
+ def test_enabled_accepts_unique_confirmed_from_owner(
+ self, sender_authentication_header):
+ user = self._identify(
+ mailfrom='attacker@example.net',
+ sender_authentication_headers=[sender_authentication_header])
+
+ assert user.username == 'test-admin'
+
+ def test_enabled_accepts_case_insensitive_canonical_header_name(self):
+ header_name = 'x-sourceforge-sender-auth'
+ user = self._identify(
+ mailfrom='attacker@example.net',
+ auth_config={
+ 'forgemail.sender_authentication.enabled': 'true',
+ 'forgemail.sender_authentication.header': header_name,
+ },
+ header_name=header_name)
+
+ assert user.username == 'test-admin'
+
+ def test_enabled_rejects_noncanonical_header_name(self):
+ user = self._identify(
+ auth_config={
+ 'forgemail.sender_authentication.enabled': 'true',
+ 'forgemail.sender_authentication.header': 'X-Other-Sender-Auth',
+ })
+
+ assert user.is_anonymous()
+
+ def test_enabled_accepts_folded_header(self):
+ user = self._identify(sender_authentication_headers=[
+ 'v=1; result=pass;\n method=spf-aligned;\n '
+ 'from-domain=users.localhost',
+ ])
+
+ assert user.username == 'test-admin'
+
+ def test_enabled_accepts_encoded_from_display_name(self):
+ user = self._identify(
+ mailfrom='attacker@example.net',
+ from_headers=[
+ '=?utf-8?q?J=C3=B6hn?= <te...@users.localhost>'])
+
+ assert user.username == 'test-admin'
+
+ @pytest.mark.parametrize('sender_authentication_headers', [
+ pytest.param([], id='missing-header'),
+ pytest.param([DMARC_PASS, DMARC_PASS], id='duplicate-header'),
+ pytest.param(
+ ['v=1; dmarc=fail; from-domain=users.localhost'],
+ id='dmarc-fail'),
+ pytest.param(
+ ['v=1; dmarc=none; from-domain=users.localhost'],
+ id='dmarc-none-is-not-a-pass'),
+ pytest.param(
+ ['v=1; result=fail; method=spf-aligned; '
+ 'from-domain=users.localhost'],
+ id='spf-fallback-fail'),
+ pytest.param(
+ ['v=1; result=pass; method=spf; '
+ 'from-domain=users.localhost'],
+ id='unsupported-method'),
+ pytest.param(
+ ['v=2; dmarc=pass; from-domain=users.localhost'],
+ id='unsupported-version'),
+ pytest.param(
+ ['dmarc=pass; v=1; from-domain=users.localhost'],
+ id='reordered-fields'),
+ pytest.param(
+ ['v=1; dmarc=pass; from-domain=users.localhost; extra=value'],
+ id='extra-field'),
+ pytest.param(
+ ['note=dmarc=pass; v=1; from-domain=users.localhost'],
+ id='dmarc-pass-substring'),
+ pytest.param(
+ ['v=1; dmarc=pass; from-domain=example.net'],
+ id='from-domain-mismatch'),
+ pytest.param(
+ ['v=1; dmarc=pass; from-domain=users.localhost\x00'],
+ id='header-control-character'),
+ pytest.param(
+ ['v=1; dmarc=pass; from-domain=' + ('a' * 1024)],
+ id='header-too-long'),
+ pytest.param(
+ ['mx.sourceforge.net; dmarc=pass '
+ 'header.from=users.localhost'],
+ id='authentication-results-shape'),
+ ])
+ def test_enabled_rejects_invalid_authentication_header(
+ self, sender_authentication_headers):
+ user = self._identify(
+ sender_authentication_headers=sender_authentication_headers)
+
+ assert user.is_anonymous()
+
+ def test_enabled_ignores_standard_authentication_results(self):
+ user = self._identify(
+ sender_authentication_headers=[],
+ extra_headers=[
+ 'Authentication-Results: mx.sourceforge.net; '
+ 'dmarc=pass header.from=users.localhost',
+ ])
+
+ assert user.is_anonymous()
+
+ @pytest.mark.parametrize('from_headers', [
+ [],
+ ['test-admin@users.localhost', 'other@users.localhost'],
+ ['test-admin@users.localhost, other@users.localhost'],
+ ['Friends: test-admin@users.localhost;'],
+ ['not-an-email-address'],
+ ])
+ def test_enabled_rejects_ambiguous_from(self, from_headers):
+ user = self._identify(from_headers=from_headers)
+
+ assert user.is_anonymous()
+
+ @pytest.mark.parametrize('owner_state', [
+ 'missing',
+ 'unconfirmed',
+ 'disabled',
+ 'pending',
+ 'duplicate',
+ ])
+ def test_enabled_requires_confirmed_unique_active_owner(self, owner_state):
+ address = 'authenticated-owner@example.net'
+ if owner_state != 'missing':
+ owner = M.User.by_username('test-user-1')
+ email_address = owner.claim_address(address)
+ email_address.confirmed = owner_state != 'unconfirmed'
+ if owner_state == 'disabled':
+ owner.disabled = True
+ elif owner_state == 'pending':
+ owner.pending = True
+ elif owner_state == 'duplicate':
+ other_owner = M.User.by_username('test-user-2')
+ other_address = other_owner.claim_address(address)
+ other_address.confirmed = True
+ ThreadLocalODMSession.flush_all()
+
+ user = self._identify(
+ from_headers=[address],
+ sender_authentication_headers=[
+ 'v=1; dmarc=pass; from-domain=example.net'])
+
+ assert user.is_anonymous()
+
+ @pytest.mark.parametrize('auth_config', [
+ {
+ 'forgemail.sender_authentication.enabled': 'unexpected',
+ 'forgemail.sender_authentication.header': HEADER_NAME,
+ },
+ {
+ 'forgemail.sender_authentication.enabled': 'true',
+ 'forgemail.sender_authentication.header': '',
+ },
+ {
+ 'forgemail.sender_authentication.enabled': 'true',
+ 'forgemail.sender_authentication.header': None,
+ },
+ {
+ 'forgemail.sender_authentication.enabled': 'true',
+ 'forgemail.sender_authentication.header': 'Not A Header Name',
+ },
+ ])
+ def test_invalid_enabled_config_fails_closed(self, auth_config):
+ user = self._identify(auth_config=auth_config)
+
+ assert user.is_anonymous()
+
+ @mock.patch(
+ 'allura.lib.mail_util._authenticated_sender',
+ side_effect=RuntimeError('verification failed unexpectedly'))
+ def test_enabled_fails_closed_on_unexpected_error(self, authenticated_sender):
+ user = self._identify()
+
+ assert user.is_anonymous()
+ authenticated_sender.assert_called_once()
+
+ @mock.patch('allura.lib.mail_util.log')
+ def test_enabled_logs_authentication_method(self, log):
+ user = self._identify(
+ sender_authentication_headers=[self.SPF_ALIGNED_PASS])
+
+ assert user.username == 'test-admin'
+ log.info.assert_called_once_with(
+ 'Inbound sender authentication enabled=true result=%s reason=%s '
+ 'peer=%s',
+ True,
+ 'spf_aligned_pass',
+ '127.0.0.1',
+ )
+
+
def test_parse_message_id():
assert _parse_message_id('<de...@libjpeg-turbo.p.domain.net>, </p...@libjpeg-turbo.p.domain.net>') == [
'de31888f6be2d87dc377d9e713876bb514548625.patches@libjpeg-turbo.p.domain.net',
diff --git a/Allura/allura/tests/test_tasks.py b/Allura/allura/tests/test_tasks.py
index 953a44ac1..9b5b3730e 100644
--- a/Allura/allura/tests/test_tasks.py
+++ b/Allura/allura/tests/test_tasks.py
@@ -507,6 +507,64 @@ def test_receive_email_ok(self):
assert args[0] == 'Page'
assert len(args) == 2
+ @td.with_wiki
+ @mock.patch.dict(tg.config, {
+ 'forgemail.sender_authentication.enabled': 'true',
+ 'forgemail.sender_authentication.header': 'X-SourceForge-Sender-Auth',
+ })
+ def test_route_email_enabled_uses_authenticated_from_user(self):
+ message = dedent('''\
+ From: "Test Admin" <te...@users.localhost>
+ X-SourceForge-Sender-Auth: v=1; dmarc=pass;
+ from-domain=users.localhost
+ Subject: authenticated inbound message
+
+ body''')
+ import forgewiki
+ senders = []
+
+ def capture_sender(*args):
+ senders.append(c.user.username)
+
+ with mock.patch.object(
+ forgewiki.wiki_main.ForgeWikiApp, 'handle_message',
+ side_effect=capture_sender):
+ mail_tasks.route_email(
+ ('127.0.0.1', 2525), 'attacker@example.net',
+ ['Page@wiki.test.p.in.localhost'], message)
+
+ assert senders == ['test-admin']
+
+ @td.with_wiki
+ @mock.patch.dict(tg.config, {
+ 'forgemail.sender_authentication.enabled': 'true',
+ 'forgemail.sender_authentication.header': 'X-SourceForge-Sender-Auth',
+ })
+ def test_route_email_enabled_does_not_trust_spoofed_envelope(self):
+ message = dedent('''\
+ From: "Test Admin" <te...@users.localhost>
+ X-SourceForge-Sender-Auth: v=1; dmarc=fail;
+ from-domain=users.localhost
+ Subject: unauthenticated inbound message
+
+ body''')
+ import forgewiki
+ senders = []
+
+ def capture_sender(*args):
+ senders.append(c.user.username)
+
+ with mock.patch.object(
+ forgewiki.wiki_main.ForgeWikiApp, 'has_access', return_value=True), \
+ mock.patch.object(
+ forgewiki.wiki_main.ForgeWikiApp, 'handle_message',
+ side_effect=capture_sender):
+ mail_tasks.route_email(
+ ('127.0.0.1', 2525), 'test-admin@users.localhost',
+ ['Page@wiki.test.p.in.localhost'], message)
+
+ assert senders == ['*anonymous']
+
@td.with_tool('test', 'Tickets', 'bugs')
def test_receive_autoresponse(self):
message = dedent('''\
diff --git a/Allura/development.ini b/Allura/development.ini
index a76316620..f5940ab83 100644
--- a/Allura/development.ini
+++ b/Allura/development.ini
@@ -631,6 +631,11 @@ forgemail.port = 8825
; domain suffix for your mail, change this. You also need to route *.*.*.forgemail.domain to the above host/port via
; your mail and DNS configuration
forgemail.domain = .in.localhost
+; When enabled, identify inbound senders only from the trusted custom header
+; added by the mail server. Keep this disabled until the mail path strips
+; inbound copies of the configured header before adding its own result.
+forgemail.sender_authentication.enabled = false
+forgemail.sender_authentication.header = X-SourceForge-Sender-Auth
; Specify the number of projects allowed to be created by a user depending on the age of their user account.
; Keys are number of seconds, values are max number allowed until that time period is reached.
(allura) 02/02: [#6054] Simplify trusted MX header verification
Posted by cc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch cc/6054-custom-header
in repository https://gitbox.apache.org/repos/asf/allura.git
commit 0ca0fbef2efae47ac23b77460071cc31a24790e7
Author: Carlos Cruz <ca...@slashdotmedia.com>
AuthorDate: Mon Aug 17 13:48:28 2026 -0600
[#6054] Simplify trusted MX header verification
---
Allura/allura/lib/mail_util.py | 244 +++++-----------------------------
Allura/allura/tests/test_mail_util.py | 241 +++++++--------------------------
2 files changed, 81 insertions(+), 404 deletions(-)
diff --git a/Allura/allura/lib/mail_util.py b/Allura/allura/lib/mail_util.py
index 06ee6f664..72436225d 100644
--- a/Allura/allura/lib/mail_util.py
+++ b/Allura/allura/lib/mail_util.py
@@ -23,8 +23,8 @@
from email.mime.text import MIMEText
from email import header
from email.message import EmailMessage
+from email.utils import parseaddr
-import idna
import six
import tg
from paste.deploy.converters import asbool, asint, aslist
@@ -32,7 +32,7 @@
from tg import tmpl_context as c
from tg import app_globals as g
-from allura.lib.utils import ConfigProxy
+from allura.lib.utils import ConfigProxy, normalize_hostname
from allura.lib import exceptions as exc
from allura.lib import helpers as h
@@ -50,16 +50,9 @@
# http://www.jebriggs.com/blog/2010/07/smtp-maximum-line-lengths/
MAX_MAIL_LINE_OCTETS = 990
-MAX_SENDER_AUTHENTICATION_HEADER_OCTETS = 1024
-MAX_FROM_HEADER_OCTETS = 2 * 1024
SENDER_AUTHENTICATION_HEADER = 'X-SourceForge-Sender-Auth'
-RE_HEADER_NAME = re.compile(r'(?a)\A[A-Za-z0-9][A-Za-z0-9-]{0,126}\Z')
-RE_DMARC_SENDER_AUTHENTICATION = re.compile(
- r'(?a)\Av=1;[ \t]+dmarc=pass;[ \t]+'
- r'from-domain=([A-Za-z0-9.-]+)\Z')
-RE_SPF_SENDER_AUTHENTICATION = re.compile(
- r'(?a)\Av=1;[ \t]+result=pass;[ \t]+method=spf-aligned;[ \t]+'
- r'from-domain=([A-Za-z0-9.-]+)\Z')
+DMARC_HEADER_PREFIX = 'v=1; dmarc=pass; from-domain='
+SPF_HEADER_PREFIX = 'v=1; result=pass; method=spf-aligned; from-domain='
email_policy = email.policy.SMTP + email.policy.strict
@@ -157,20 +150,6 @@ def parse_message(data):
# Extract relevant data
result = {}
result['multipart'] = multipart = msg.is_multipart()
- # dict(msg) intentionally retains the longstanding, first-header-wins view
- # used by message handlers. Sender authentication separately preserves all
- # From and configured authentication fields so duplicates fail closed.
- raw_headers = list(msg.raw_items())
- result['from_headers'] = [
- value for name, value in raw_headers if name.casefold() == 'from']
- authentication_header = config.get('sender_authentication_header', '')
- if isinstance(authentication_header, str):
- authentication_header = authentication_header.strip().casefold()
- else:
- authentication_header = ''
- result['sender_authentication_headers'] = [
- value for name, value in raw_headers
- if authentication_header and name.casefold() == authentication_header]
result['headers'] = dict(msg)
result['message_id'] = _parse_message_id(msg.get('Message-ID'))
result['in_reply_to'] = _parse_message_id(msg.get('In-Reply-To'))
@@ -203,176 +182,34 @@ def parse_message(data):
return result
-class SenderAuthenticationError(ValueError):
-
- def __init__(self, reason):
- self.reason = reason
- super().__init__(reason)
-
+def _authenticated_sender(msg):
+ from allura import model as M
-def _unfold_sender_authentication_header(value, max_octets):
- if not isinstance(value, str):
- raise SenderAuthenticationError('header_type')
- try:
- if len(value.encode('utf-8')) > max_octets:
- raise SenderAuthenticationError('header_too_long')
- except UnicodeEncodeError:
- raise SenderAuthenticationError('header_unicode')
-
- unfolded = []
- i = 0
- while i < len(value):
- char = value[i]
- if char in '\r\n':
- if char == '\r':
- if i + 1 >= len(value) or value[i + 1] != '\n':
- raise SenderAuthenticationError('header_bad_fold')
- i += 2
- else:
- i += 1
- if i >= len(value) or value[i] not in ' \t':
- raise SenderAuthenticationError('header_bad_fold')
- while i < len(value) and value[i] in ' \t':
- i += 1
- unfolded.append(' ')
- continue
- codepoint = ord(char)
- if ((codepoint < 0x20 and char != '\t') or codepoint == 0x7f or
- 0x80 <= codepoint <= 0x9f):
- raise SenderAuthenticationError('header_control')
- unfolded.append(char)
- i += 1
-
- value = ''.join(unfolded).strip(' \t')
- if not value:
- raise SenderAuthenticationError('header_empty')
- return value
-
-
-def _normalize_sender_domain(domain):
- if (not isinstance(domain, str) or not domain or
- domain.startswith('.') or domain.endswith('.')):
- raise SenderAuthenticationError('domain_invalid')
- try:
- normalized = idna.encode(
- domain,
- uts46=True,
- transitional=False,
- std3_rules=True,
- ).decode('ascii').lower()
- except (idna.IDNAError, UnicodeError):
- raise SenderAuthenticationError('domain_invalid')
- labels = normalized.split('.')
- if (len(normalized) > 253 or '..' in normalized or
- any(not label or len(label) > 63 for label in labels)):
- raise SenderAuthenticationError('domain_invalid')
- return normalized
-
-
-def _parse_sender_authentication_header(value, from_domain):
- value = _unfold_sender_authentication_header(
- value, MAX_SENDER_AUTHENTICATION_HEADER_OCTETS)
- for method, pattern in (
- ('dmarc', RE_DMARC_SENDER_AUTHENTICATION),
- ('spf_aligned', RE_SPF_SENDER_AUTHENTICATION)):
- match = pattern.fullmatch(value)
- if match:
+ anonymous = M.User.anonymous()
+ headers = msg.get('headers', {}) if isinstance(msg, dict) else {}
+ mailbox = parseaddr(headers.get('From', ''))[1]
+ from_domain = normalize_hostname(mailbox.rpartition('@')[2])
+ header_name = config.get(
+ 'sender_authentication_header', SENDER_AUTHENTICATION_HEADER)
+ value = ' '.join(headers.get(header_name, '').split())
+
+ for method, prefix in (
+ ('dmarc', DMARC_HEADER_PREFIX),
+ ('spf_aligned', SPF_HEADER_PREFIX)):
+ if value.startswith(prefix):
+ authenticated_domain = value.removeprefix(prefix)
break
else:
- raise SenderAuthenticationError('sender_authentication_syntax')
-
- if _normalize_sender_domain(match.group(1)) != from_domain:
- raise SenderAuthenticationError('from_domain_mismatch')
- return method
-
-
-def _parse_from_mailbox(value):
- value = _unfold_sender_authentication_header(
- value, MAX_FROM_HEADER_OCTETS)
- parsed = email.parser.HeaderParser(policy=email.policy.default).parsestr(
- f'From: {value}\n\n')
- field = parsed['From']
- if field is None or field.defects:
- raise SenderAuthenticationError('from_defect')
- addresses = tuple(field.addresses)
- groups = tuple(field.groups)
- if (len(addresses) != 1 or
- any(group.display_name is not None for group in groups)):
- raise SenderAuthenticationError('from_mailbox_count')
- address = addresses[0]
- if not address.username or not address.domain:
- raise SenderAuthenticationError('from_invalid')
- mailbox = address.addr_spec
- if len(mailbox.encode('utf-8')) > 320:
- raise SenderAuthenticationError('from_too_long')
- return mailbox, _normalize_sender_domain(address.domain)
-
-
-def _configured_sender_authentication_header():
- header_name = config.get('sender_authentication_header', '')
- if not isinstance(header_name, str):
- raise SenderAuthenticationError('sender_authentication_header_config')
- header_name = header_name.strip()
- # Exim strips this reserved name before adding its own attestation. Treat
- # the setting as a deployment-parity check; accepting an arbitrary name
- # would let an inbound copy under that other name bypass Exim's stripping.
- if (not RE_HEADER_NAME.fullmatch(header_name) or
- header_name.casefold() !=
- SENDER_AUTHENTICATION_HEADER.casefold()):
- raise SenderAuthenticationError('sender_authentication_header_config')
- return header_name
-
-
-def _sender_authentication_enabled():
- value = config.get('sender_authentication_enabled', False)
- if isinstance(value, bool):
- return value
- if isinstance(value, str):
- value = value.strip().lower()
- if value in {'true', 'yes', 'on', '1'}:
- return True
- if value in {'false', 'no', 'off', '0'}:
- return False
- raise ValueError('invalid sender authentication enabled value')
+ return anonymous, 'authentication_failed'
+ if (not from_domain or
+ normalize_hostname(authenticated_domain) != from_domain):
+ return anonymous, 'from_domain_mismatch'
-def _authenticated_sender(msg):
- from allura import model as M
-
- anonymous = M.User.anonymous()
- try:
- _configured_sender_authentication_header()
- if not isinstance(msg, dict):
- raise SenderAuthenticationError('message_missing')
- authentication_headers = msg.get('sender_authentication_headers')
- from_headers = msg.get('from_headers')
- if (not isinstance(authentication_headers, (list, tuple)) or
- len(authentication_headers) != 1):
- raise SenderAuthenticationError('sender_authentication_header_count')
- if (not isinstance(from_headers, (list, tuple)) or
- len(from_headers) != 1):
- raise SenderAuthenticationError('from_count')
-
- mailbox, from_domain = _parse_from_mailbox(from_headers[0])
- authentication_method = _parse_sender_authentication_header(
- authentication_headers[0], from_domain)
-
- # DMARC or aligned SPF proves control of the From domain, not a
- # cryptographic per-mailbox identity. The exact confirmed address is
- # the application-side mapping for that domain-authenticated mailbox.
- owners = {}
- for address in M.EmailAddress.find({
- 'email': mailbox,
- 'confirmed': True,
- }):
- owner = address.claimed_by_user()
- if owner is not None:
- owners[owner._id] = owner
- if len(owners) != 1:
- raise SenderAuthenticationError('active_owner_count')
- return next(iter(owners.values())), f'{authentication_method}_pass'
- except SenderAuthenticationError as error:
- return anonymous, error.reason
+ user = M.User.by_email_address(mailbox)
+ if user is None:
+ return anonymous, 'owner_not_found'
+ return user, f'{method}_pass'
def _legacy_identify_sender(email_address, headers):
@@ -391,34 +228,13 @@ def _legacy_identify_sender(email_address, headers):
def identify_sender(peer, email_address, headers, msg):
- from allura import model as M
-
- try:
- enabled = _sender_authentication_enabled()
- except ValueError:
- log.error(
- 'Invalid forgemail sender authentication enabled value; '
- 'using anonymous sender')
- return M.User.anonymous()
-
- if not enabled:
+ if not asbool(config.get('sender_authentication_enabled', False)):
return _legacy_identify_sender(email_address, headers)
- try:
- verified_user, reason = _authenticated_sender(msg)
- except Exception:
- log.exception('Unexpected inbound sender authentication error')
- return M.User.anonymous()
- peer_address = (
- peer[0]
- if isinstance(peer, (tuple, list)) and peer
- else 'invalid')
+ verified_user, reason = _authenticated_sender(msg)
log.info(
'Inbound sender authentication enabled=true result=%s reason=%s '
'peer=%s',
- not verified_user.is_anonymous(),
- reason,
- peer_address,
- )
+ not verified_user.is_anonymous(), reason, peer[0])
return verified_user
diff --git a/Allura/allura/tests/test_mail_util.py b/Allura/allura/tests/test_mail_util.py
index 05b4b1c59..3e1d95ee2 100644
--- a/Allura/allura/tests/test_mail_util.py
+++ b/Allura/allura/tests/test_mail_util.py
@@ -213,26 +213,6 @@ def test_unicode_complex_message(self):
continue
assert isinstance(part['payload'], str), type(part['payload'])
- @mock.patch.dict(tg_config, {
- 'forgemail.sender_authentication.header': 'X-Test-Sender-Auth',
- })
- def test_parse_message_preserves_repeated_sender_authentication_headers(self):
- msg = parse_message('''\
-From: first@example.com
-From: second@example.com
-X-Test-Sender-Auth: v=1; dmarc=pass; from-domain=example.com
-x-test-sender-auth:
- v=1; result=pass; method=spf-aligned; from-domain=example.com
-
-body''')
-
- assert msg['from_headers'] == ['first@example.com', 'second@example.com']
- assert msg['sender_authentication_headers'] == [
- 'v=1; dmarc=pass; from-domain=example.com',
- '\n v=1; result=pass; method=spf-aligned; from-domain=example.com',
- ]
-
-
class TestHeader:
def test_bytestring(self):
@@ -358,39 +338,31 @@ def setup_method(self, method):
setup_global_objects()
ThreadLocalODMSession.flush_all()
- def _message(self, from_headers=None, sender_authentication_headers=None,
- header_name=None, extra_headers=None):
- if from_headers is None:
- from_headers = ['"Test Admin" <te...@users.localhost>']
- if sender_authentication_headers is None:
- sender_authentication_headers = [self.DMARC_PASS]
- if header_name is None:
- header_name = self.HEADER_NAME
- lines = []
- lines.extend(f'From: {value}' for value in from_headers)
- lines.extend(
- f'{header_name}: {value}'
- for value in sender_authentication_headers)
- lines.extend(extra_headers or [])
+ def _message(self, from_header=None, sender_authentication_header=None):
+ if from_header is None:
+ from_header = '"Test Admin" <te...@users.localhost>'
+ if sender_authentication_header is None:
+ sender_authentication_header = self.DMARC_PASS
+ lines = [f'From: {from_header}']
+ if sender_authentication_header:
+ lines.append(
+ f'{self.HEADER_NAME}: {sender_authentication_header}')
lines.extend(['Subject: inbound sender authentication test', '', 'body'])
return parse_message('\n'.join(lines))
def _identify(self, *, peer=('127.0.0.1', 2525),
- mailfrom='test-admin@users.localhost', from_headers=None,
- sender_authentication_headers=None, auth_config=None,
- header_name=None, extra_headers=None):
+ mailfrom='test-admin@users.localhost', from_header=None,
+ sender_authentication_header=None, auth_config=None):
if auth_config is None:
auth_config = self.AUTH_CONFIG
with mock.patch.dict(tg_config, auth_config):
- msg = self._message(
- from_headers, sender_authentication_headers, header_name,
- extra_headers)
+ msg = self._message(from_header, sender_authentication_header)
return identify_sender(peer, mailfrom, msg['headers'], msg)
def test_disabled_preserves_envelope_sender_lookup(self):
user = self._identify(
peer=('203.0.113.10', 2525),
- sender_authentication_headers=[],
+ sender_authentication_header='',
auth_config={
'forgemail.sender_authentication.enabled': 'false',
'forgemail.sender_authentication.header': self.HEADER_NAME,
@@ -403,189 +375,78 @@ def test_disabled_preserves_envelope_sender_lookup(self):
SPF_ALIGNED_PASS,
'v=1; dmarc=pass; from-domain=USERS.LOCALHOST',
])
- def test_enabled_accepts_unique_confirmed_from_owner(
+ def test_enabled_accepts_confirmed_from_owner(
self, sender_authentication_header):
user = self._identify(
mailfrom='attacker@example.net',
- sender_authentication_headers=[sender_authentication_header])
-
- assert user.username == 'test-admin'
-
- def test_enabled_accepts_case_insensitive_canonical_header_name(self):
- header_name = 'x-sourceforge-sender-auth'
- user = self._identify(
- mailfrom='attacker@example.net',
- auth_config={
- 'forgemail.sender_authentication.enabled': 'true',
- 'forgemail.sender_authentication.header': header_name,
- },
- header_name=header_name)
+ sender_authentication_header=sender_authentication_header)
assert user.username == 'test-admin'
- def test_enabled_rejects_noncanonical_header_name(self):
- user = self._identify(
- auth_config={
- 'forgemail.sender_authentication.enabled': 'true',
- 'forgemail.sender_authentication.header': 'X-Other-Sender-Auth',
- })
-
- assert user.is_anonymous()
-
def test_enabled_accepts_folded_header(self):
- user = self._identify(sender_authentication_headers=[
+ user = self._identify(sender_authentication_header=(
'v=1; result=pass;\n method=spf-aligned;\n '
- 'from-domain=users.localhost',
- ])
+ 'from-domain=users.localhost'))
assert user.username == 'test-admin'
def test_enabled_accepts_encoded_from_display_name(self):
user = self._identify(
mailfrom='attacker@example.net',
- from_headers=[
- '=?utf-8?q?J=C3=B6hn?= <te...@users.localhost>'])
+ from_header=(
+ '=?utf-8?q?J=C3=B6hn?= <te...@users.localhost>'))
assert user.username == 'test-admin'
- @pytest.mark.parametrize('sender_authentication_headers', [
- pytest.param([], id='missing-header'),
- pytest.param([DMARC_PASS, DMARC_PASS], id='duplicate-header'),
- pytest.param(
- ['v=1; dmarc=fail; from-domain=users.localhost'],
- id='dmarc-fail'),
- pytest.param(
- ['v=1; dmarc=none; from-domain=users.localhost'],
- id='dmarc-none-is-not-a-pass'),
- pytest.param(
- ['v=1; result=fail; method=spf-aligned; '
- 'from-domain=users.localhost'],
- id='spf-fallback-fail'),
- pytest.param(
- ['v=1; result=pass; method=spf; '
- 'from-domain=users.localhost'],
- id='unsupported-method'),
- pytest.param(
- ['v=2; dmarc=pass; from-domain=users.localhost'],
- id='unsupported-version'),
- pytest.param(
- ['dmarc=pass; v=1; from-domain=users.localhost'],
- id='reordered-fields'),
- pytest.param(
- ['v=1; dmarc=pass; from-domain=users.localhost; extra=value'],
- id='extra-field'),
- pytest.param(
- ['note=dmarc=pass; v=1; from-domain=users.localhost'],
- id='dmarc-pass-substring'),
- pytest.param(
- ['v=1; dmarc=pass; from-domain=example.net'],
- id='from-domain-mismatch'),
- pytest.param(
- ['v=1; dmarc=pass; from-domain=users.localhost\x00'],
- id='header-control-character'),
- pytest.param(
- ['v=1; dmarc=pass; from-domain=' + ('a' * 1024)],
- id='header-too-long'),
- pytest.param(
- ['mx.sourceforge.net; dmarc=pass '
- 'header.from=users.localhost'],
- id='authentication-results-shape'),
- ])
- def test_enabled_rejects_invalid_authentication_header(
- self, sender_authentication_headers):
- user = self._identify(
- sender_authentication_headers=sender_authentication_headers)
-
- assert user.is_anonymous()
+ def test_enabled_accepts_equivalent_idna_domains(self):
+ owner = M.User.by_username('test-user-1')
+ address = owner.claim_address('idn@xn--bcher-kva.example')
+ address.confirmed = True
+ ThreadLocalODMSession.flush_all()
+ msg = {'headers': {
+ 'From': 'idn@xn--bcher-kva.example',
+ self.HEADER_NAME: (
+ 'v=1; dmarc=pass; from-domain=bücher.example'),
+ }}
- def test_enabled_ignores_standard_authentication_results(self):
- user = self._identify(
- sender_authentication_headers=[],
- extra_headers=[
- 'Authentication-Results: mx.sourceforge.net; '
- 'dmarc=pass header.from=users.localhost',
- ])
+ with mock.patch.dict(tg_config, self.AUTH_CONFIG):
+ user = identify_sender(
+ ('127.0.0.1', 2525), 'attacker@example.net',
+ msg['headers'], msg)
- assert user.is_anonymous()
+ assert user.username == 'test-user-1'
- @pytest.mark.parametrize('from_headers', [
- [],
- ['test-admin@users.localhost', 'other@users.localhost'],
- ['test-admin@users.localhost, other@users.localhost'],
- ['Friends: test-admin@users.localhost;'],
- ['not-an-email-address'],
+ @pytest.mark.parametrize('sender_authentication_header', [
+ 'v=1; dmarc=fail; from-domain=users.localhost',
+ 'v=1; result=fail; method=spf-aligned; '
+ 'from-domain=users.localhost',
+ 'v=1; dmarc=pass; from-domain=example.net',
])
- def test_enabled_rejects_ambiguous_from(self, from_headers):
- user = self._identify(from_headers=from_headers)
+ def test_enabled_rejects_failed_or_misaligned_authentication(
+ self, sender_authentication_header):
+ user = self._identify(
+ sender_authentication_header=sender_authentication_header)
assert user.is_anonymous()
- @pytest.mark.parametrize('owner_state', [
- 'missing',
- 'unconfirmed',
- 'disabled',
- 'pending',
- 'duplicate',
- ])
- def test_enabled_requires_confirmed_unique_active_owner(self, owner_state):
+ def test_enabled_requires_confirmed_active_owner(self):
address = 'authenticated-owner@example.net'
- if owner_state != 'missing':
- owner = M.User.by_username('test-user-1')
- email_address = owner.claim_address(address)
- email_address.confirmed = owner_state != 'unconfirmed'
- if owner_state == 'disabled':
- owner.disabled = True
- elif owner_state == 'pending':
- owner.pending = True
- elif owner_state == 'duplicate':
- other_owner = M.User.by_username('test-user-2')
- other_address = other_owner.claim_address(address)
- other_address.confirmed = True
+ owner = M.User.by_username('test-user-1')
+ email_address = owner.claim_address(address)
+ email_address.confirmed = False
ThreadLocalODMSession.flush_all()
user = self._identify(
- from_headers=[address],
- sender_authentication_headers=[
- 'v=1; dmarc=pass; from-domain=example.net'])
-
- assert user.is_anonymous()
-
- @pytest.mark.parametrize('auth_config', [
- {
- 'forgemail.sender_authentication.enabled': 'unexpected',
- 'forgemail.sender_authentication.header': HEADER_NAME,
- },
- {
- 'forgemail.sender_authentication.enabled': 'true',
- 'forgemail.sender_authentication.header': '',
- },
- {
- 'forgemail.sender_authentication.enabled': 'true',
- 'forgemail.sender_authentication.header': None,
- },
- {
- 'forgemail.sender_authentication.enabled': 'true',
- 'forgemail.sender_authentication.header': 'Not A Header Name',
- },
- ])
- def test_invalid_enabled_config_fails_closed(self, auth_config):
- user = self._identify(auth_config=auth_config)
-
- assert user.is_anonymous()
-
- @mock.patch(
- 'allura.lib.mail_util._authenticated_sender',
- side_effect=RuntimeError('verification failed unexpectedly'))
- def test_enabled_fails_closed_on_unexpected_error(self, authenticated_sender):
- user = self._identify()
+ from_header=address,
+ sender_authentication_header=(
+ 'v=1; dmarc=pass; from-domain=example.net'))
assert user.is_anonymous()
- authenticated_sender.assert_called_once()
@mock.patch('allura.lib.mail_util.log')
def test_enabled_logs_authentication_method(self, log):
user = self._identify(
- sender_authentication_headers=[self.SPF_ALIGNED_PASS])
+ sender_authentication_header=self.SPF_ALIGNED_PASS)
assert user.username == 'test-admin'
log.info.assert_called_once_with(