Merge commit from fork

With [smtp] ssl = False (the default) or [imap] ssl = False, mailsuite
upgraded the connection only if the server's plain-text EHLO/CAPABILITY
reply advertised STARTTLS, and otherwise authenticated in plain text. An
on-path attacker who stripped STARTTLS received the credentials and the
emailed results (CWE-319).

- Require mailsuite >= 2.3.4 (GHSA-2x4j-7g3q-wr6f), whose default
  refuses to authenticate over a connection that was not upgraded.
- Add [imap] and [smtp] starttls (unset = require STARTTLS; false =
  explicit plain-text opt-out, for loopback servers such as DavMail),
  with a startup warning when the opt-out is set and ssl is False.
- Add a starttls argument to email_results().
- Document both options; set starttls = False in the DavMail example.

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Sean WhalenandClaude Opus 5.5 authored and GitHub committed 2026-10-31 20:36:09 -04:00
1 parent dc3b14dc0e
commit 33ca4ecf71
8 files changed
+590 -9

No files matched your search

+281
View File
@@ -2965,6 +2965,287 @@ client_assertion = s3cret-signed-jwt-assertion
self.assertEqual(logging.getLogger(name).level, logging.WARNING, name)
def _recording_smtp_class(kind: str, calls: list, *, advertise_starttls: bool):
"""A stand-in for ``smtplib.SMTP`` / ``smtplib.SMTP_SSL`` that appends
each call it receives to ``calls`` in order. ``advertise_starttls=False``
plays a server, or an on-path attacker, whose EHLO reply has no
STARTTLS (GHSA-3588-2mr7-vrcm)."""
class _RecordingSMTP:
def __init__(self, host="", port=0, *args, **kwargs):
calls.append((kind, host, port))
def ehlo_or_helo_if_needed(self):
calls.append(("ehlo",))
def ehlo(self):
calls.append(("ehlo",))
def has_extn(self, name):
return name.lower() == "starttls" and advertise_starttls
def starttls(self, context=None):
calls.append(("starttls",))
def login(self, user, password):
calls.append(("login", user, password))
def sendmail(self, sender, to, msg):
calls.append(("sendmail", sender, sorted(to), msg))
return _RecordingSMTP
class TestCliStartTLS(unittest.TestCase):
"""GHSA-3588-2mr7-vrcm: the ``[smtp]`` and ``[imap]`` ``starttls``
option. Unset, a plain-text (``ssl = False``) connection must be
upgraded with STARTTLS before credentials are sent; ``starttls =
false`` is the explicit plain-text opt-out and is logged as a WARNING.
SMTP is exercised end to end through ``_main()`` with ``smtplib``
patched; IMAP through the kwargs handed to ``IMAPConnection``."""
PASSWORD = "Sup3rSecret-placeholder"
SMTP_CONFIG = """[general]
offline = true
silent = true
[smtp]
host = smtp.example.com
port = 587
user = dmarc@example.com
password = Sup3rSecret-placeholder
from = dmarc@example.com
to = admin@example.com
"""
IMAP_CONFIG = """[general]
offline = true
silent = true
[imap]
host = imap.example.com
port = 143
ssl = false
user = dmarc@example.com
password = Sup3rSecret-placeholder
"""
def setUp(self):
# Every SMTP test parses the same sample report; without a fresh
# dedup cache the second run would skip it as already seen, parse
# nothing, and skip the results email.
parsedmarc.SEEN_AGGREGATE_REPORT_IDS.clear()
self.addCleanup(parsedmarc.SEEN_AGGREGATE_REPORT_IDS.clear)
# _main() reconfigures process-global dependency loggers; restore
# them so these tests don't leak state into the rest of the suite.
saved = {}
for name in parsedmarc.cli._DEPENDENCY_LOGGERS:
dep = logging.getLogger(name)
saved[name] = (dep.level, list(dep.handlers), dep.propagate)
def restore():
for name, (level, handlers, propagate) in saved.items():
dep = logging.getLogger(name)
dep.setLevel(level)
dep.handlers = handlers
dep.propagate = propagate
self.addCleanup(restore)
def _write_config(self, config_text):
with tempfile.NamedTemporaryFile("w", suffix=".ini", delete=False) as cfg:
cfg.write(config_text)
cfg_path = cfg.name
self.addCleanup(lambda: os.path.exists(cfg_path) and os.remove(cfg_path))
return cfg_path
def _capture_records(self) -> list[logging.LogRecord]:
"""Collect every record the parsedmarc logger emits from here on.
The level is lowered to WARNING (as ``assertLogs`` does in the
positive tests) because ``_parse_config()`` runs before ``_main()``
sets the level, and an earlier ``_main()`` in this process may have
left it at ERROR -- which would make a "no WARNING" assertion pass
vacuously."""
records: list[logging.LogRecord] = []
handler = logging.Handler()
handler.emit = records.append # pyright: ignore[reportAttributeAccessIssue]
log = parsedmarc.log.logger
previous_level = log.level
log.setLevel(logging.WARNING)
log.addHandler(handler)
self.addCleanup(log.setLevel, previous_level)
self.addCleanup(log.removeHandler, handler)
return records
def _run_smtp(self, config_text, *, advertise_starttls):
calls: list = []
cfg_path = self._write_config(config_text)
argv = ["parsedmarc", "-c", cfg_path, SAMPLE_AGGREGATE_REPORT_PATH]
with (
patch(
"smtplib.SMTP",
_recording_smtp_class(
"SMTP", calls, advertise_starttls=advertise_starttls
),
),
patch(
"smtplib.SMTP_SSL",
_recording_smtp_class(
"SMTP_SSL", calls, advertise_starttls=advertise_starttls
),
),
patch.object(sys, "argv", argv),
):
parsedmarc.cli._main()
return calls
def testDefaultSmtpRefusesPlainTextAuthentication(self):
"""GHSA-3588-2mr7-vrcm: with the default ``[smtp]`` settings
(``ssl = False``, no ``starttls``) and a server whose EHLO reply has
no STARTTLS, the CLI logs "Failed to email results" and exits 1
without sending LOGIN or the report, and the password appears
nowhere in the logged error or traceback."""
calls: list = []
with self.assertLogs("parsedmarc.log", level="ERROR") as logs:
with self.assertRaises(SystemExit) as system_exit:
calls = self._run_smtp(self.SMTP_CONFIG, advertise_starttls=False)
self.assertEqual(system_exit.exception.code, 1)
output = "\n".join(logs.output)
self.assertIn("Failed to email results", output)
self.assertIn("STARTTLS", output)
self.assertNotIn(self.PASSWORD, output)
names = [call[0] for call in calls]
self.assertNotIn("login", names)
self.assertNotIn("sendmail", names)
def testDefaultSmtpUpgradesBeforeLogin(self):
"""With the default settings and a server that advertises STARTTLS,
the connection is upgraded before LOGIN and the report is sent, and
no plain-text WARNING is logged (GHSA-3588-2mr7-vrcm)."""
records = self._capture_records()
calls = self._run_smtp(self.SMTP_CONFIG, advertise_starttls=True)
names = [call[0] for call in calls]
self.assertEqual(calls[0], ("SMTP", "smtp.example.com", 587))
self.assertLess(names.index("starttls"), names.index("login"))
self.assertLess(names.index("login"), names.index("sendmail"))
self.assertEqual(
calls[names.index("login")],
("login", "dmarc@example.com", self.PASSWORD),
)
self.assertFalse(any("plain text" in record.getMessage() for record in records))
def testSmtpStartTLSFalseOptsOutWithWarning(self):
"""``[smtp] starttls = false`` reaches mailsuite: LOGIN and the report
are sent without STARTTLS, and a WARNING says the credentials go out
in plain text, without naming them (GHSA-3588-2mr7-vrcm)."""
with self.assertLogs("parsedmarc.log", level="WARNING") as logs:
calls = self._run_smtp(
self.SMTP_CONFIG + "starttls = false\n", advertise_starttls=False
)
names = [call[0] for call in calls]
self.assertNotIn("starttls", names)
self.assertEqual(
calls[names.index("login")],
("login", "dmarc@example.com", self.PASSWORD),
)
sendmail = calls[names.index("sendmail")]
self.assertEqual(sendmail[2], ["admin@example.com"])
output = "\n".join(logs.output)
self.assertIn("WARNING:parsedmarc.log:[smtp] starttls = False", output)
self.assertIn("SMTP credentials", output)
self.assertNotIn(self.PASSWORD, output)
def testSmtpStartTLSFalseWithSslLogsNoWarning(self):
"""With ``ssl = true`` the ``starttls`` option is ignored, so no
plain-text WARNING is logged and implicit TLS is used
(GHSA-3588-2mr7-vrcm)."""
records = self._capture_records()
calls = self._run_smtp(
self.SMTP_CONFIG + "ssl = true\nstarttls = false\n",
advertise_starttls=False,
)
names = [call[0] for call in calls]
self.assertEqual(names[0], "SMTP_SSL")
self.assertNotIn("starttls", names)
self.assertIn("sendmail", names)
self.assertFalse(any("plain text" in record.getMessage() for record in records))
def _run_imap(self, config_text, env=None):
cfg_path = self._write_config(config_text)
with (
patch("parsedmarc.cli.IMAPConnection") as mock_imap,
patch("parsedmarc.cli.get_dmarc_reports_from_mailbox") as mock_get,
patch.dict(os.environ, env or {}),
patch.object(sys, "argv", ["parsedmarc", "-c", cfg_path]),
):
mock_imap.return_value = object()
mock_get.return_value = {
"aggregate_reports": [],
"failure_reports": [],
"smtp_tls_reports": [],
}
parsedmarc.cli._main()
return mock_imap.call_args.kwargs
def testImapStartTLSDefaultsToNone(self):
"""Without ``[imap] starttls``, ``IMAPConnection`` gets
``starttls=None`` -- mailsuite's default, which refuses to log in
over a connection that was not upgraded (GHSA-3588-2mr7-vrcm) --
and no plain-text WARNING is logged."""
records = self._capture_records()
kwargs = self._run_imap(self.IMAP_CONFIG)
self.assertIn("starttls", kwargs)
self.assertIsNone(kwargs["starttls"])
self.assertFalse(kwargs["ssl"])
self.assertFalse(any("plain text" in record.getMessage() for record in records))
def testImapStartTLSFalseFromIni(self):
"""``[imap] starttls = false`` with ``ssl = false`` hands
``starttls=False`` to ``IMAPConnection`` and logs the plain-text
WARNING (GHSA-3588-2mr7-vrcm)."""
with self.assertLogs("parsedmarc.log", level="WARNING") as logs:
kwargs = self._run_imap(self.IMAP_CONFIG + "starttls = false\n")
self.assertIs(kwargs["starttls"], False)
output = "\n".join(logs.output)
self.assertIn("WARNING:parsedmarc.log:[imap] starttls = False", output)
self.assertIn("IMAP credentials", output)
self.assertNotIn(self.PASSWORD, output)
def testImapStartTLSFromEnvVar(self):
"""``PARSEDMARC_IMAP_STARTTLS`` reaches ``IMAPConnection`` through
the generic section/key mapping, in both directions
(GHSA-3588-2mr7-vrcm)."""
for value, expected in (("false", False), ("true", True)):
with self.subTest(value=value):
kwargs = self._run_imap(
self.IMAP_CONFIG, env={"PARSEDMARC_IMAP_STARTTLS": value}
)
self.assertIs(kwargs["starttls"], expected)
def testImapStartTLSFalseWithSslLogsNoWarning(self):
"""With ``[imap] ssl = true`` the ``starttls`` option is ignored by
mailsuite, so no plain-text WARNING is logged
(GHSA-3588-2mr7-vrcm)."""
records = self._capture_records()
kwargs = self._run_imap(
self.IMAP_CONFIG.replace("ssl = false", "ssl = true") + "starttls = false\n"
)
self.assertTrue(kwargs["ssl"])
self.assertIs(kwargs["starttls"], False)
self.assertFalse(any("plain text" in record.getMessage() for record in records))
class TestMSGraphEmailResults(unittest.TestCase):
"""#472: the periodic summary email is sent via the same
already-authenticated Microsoft Graph mailbox connection when
+218
View File
@@ -18,6 +18,7 @@ import mailbox
import os
import quopri
import time
import traceback
import unittest
from collections.abc import Callable
from datetime import datetime, timedelta, timezone
@@ -5799,6 +5800,223 @@ class TestEmailResultsErrorBranches(unittest.TestCase):
)
def _fake_smtp_class(kind: str, calls: list, *, advertise_starttls: bool) -> type:
"""Build a stand-in for ``smtplib.SMTP`` / ``smtplib.SMTP_SSL`` that
appends every call it receives to ``calls``, in order, so a test can
assert what happened on the wire and in which order (modeled on the fake
in mailsuite's ``tests/test_smtp.py``). ``kind`` names the patched class
in the constructor entry, and ``advertise_starttls`` decides whether the
EHLO reply lists STARTTLS -- ``False`` plays a server, or an on-path
attacker, that strips it (GHSA-3588-2mr7-vrcm)."""
class _FakeSMTP:
def __init__(self, host="", port=0, *args, **kwargs):
calls.append((kind, host, port))
def ehlo_or_helo_if_needed(self):
calls.append(("ehlo",))
def ehlo(self):
calls.append(("ehlo",))
def has_extn(self, name):
return name.lower() == "starttls" and advertise_starttls
def starttls(self, context=None):
calls.append(("starttls",))
def login(self, user, password):
calls.append(("login", user, password))
def auth(self, mechanism, authobject, **kwargs):
calls.append(("auth", mechanism))
def sendmail(self, sender, to, msg):
calls.append(("sendmail", sender, sorted(to), msg))
return _FakeSMTP
class TestEmailResultsStartTLS(unittest.TestCase):
"""GHSA-3588-2mr7-vrcm: with ``require_encryption=False``,
``email_results()`` used to authenticate in plain text whenever the
server's EHLO reply did not list STARTTLS, so an on-path attacker who
stripped STARTTLS from that reply received the SMTP credentials and the
report. With mailsuite >= 2.3.4 the default (``starttls=None``) refuses
to send credentials over a connection that was not upgraded, and
``starttls=False`` is the explicit plain-text opt-out. These tests patch
``smtplib`` itself (the SDK boundary under mailsuite's ``send_email()``)
and assert on the recorded calls and their payloads."""
PASSWORD = "Sup3rSecret-placeholder"
def _send(self, *, advertise_starttls: bool, **kwargs) -> list:
calls: list = []
with (
patch(
"smtplib.SMTP",
_fake_smtp_class("SMTP", calls, advertise_starttls=advertise_starttls),
),
patch(
"smtplib.SMTP_SSL",
_fake_smtp_class(
"SMTP_SSL", calls, advertise_starttls=advertise_starttls
),
),
):
try:
parsedmarc.email_results(
{
"aggregate_reports": [],
"failure_reports": [],
"smtp_tls_reports": [],
},
"smtp.example.com",
"dmarc@example.com",
["admin@example.com"],
port=587,
**kwargs,
)
finally:
self.calls = calls
return calls
def _names(self, calls: list) -> list:
return [call[0] for call in calls]
def testCredentialsWithoutStartTLSAreRefused(self):
"""GHSA-3588-2mr7-vrcm: by default, credentials are never sent when
the server does not advertise STARTTLS. The send fails with
mailsuite's SMTPError before LOGIN, nothing is sent, and the
password appears nowhere in the error or its traceback."""
from mailsuite.smtp import SMTPError
with self.assertRaises(SMTPError) as ctx:
self._send(
advertise_starttls=False,
username="dmarc@example.com",
password=self.PASSWORD,
)
names = self._names(self.calls)
self.assertEqual(names[0], "SMTP")
self.assertNotIn("SMTP_SSL", names)
self.assertNotIn("login", names)
self.assertNotIn("auth", names)
self.assertNotIn("sendmail", names)
self.assertIn("STARTTLS", str(ctx.exception))
formatted = "".join(
traceback.format_exception(
type(ctx.exception), ctx.exception, ctx.exception.__traceback__
)
)
self.assertNotIn(self.PASSWORD, formatted)
def testStartTLSPrecedesLoginAndSend(self):
"""When the server advertises STARTTLS, the connection is upgraded
before LOGIN, LOGIN gets the configured credentials, and the report
is sent last (GHSA-3588-2mr7-vrcm)."""
calls = self._send(
advertise_starttls=True,
username="dmarc@example.com",
password=self.PASSWORD,
)
names = self._names(calls)
self.assertEqual(calls[0], ("SMTP", "smtp.example.com", 587))
self.assertLess(names.index("starttls"), names.index("login"))
self.assertLess(names.index("login"), names.index("sendmail"))
login = calls[names.index("login")]
self.assertEqual(login, ("login", "dmarc@example.com", self.PASSWORD))
sendmail = calls[names.index("sendmail")]
self.assertEqual(sendmail[1], "dmarc@example.com")
self.assertEqual(sendmail[2], ["admin@example.com"])
self.assertIn("Subject: DMARC results for ", sendmail[3])
self.assertIn('filename="DMARC-', sendmail[3])
def testRequireEncryptionUsesImplicitTLS(self):
"""``require_encryption=True`` connects with ``SMTP_SSL`` and never
issues STARTTLS, whatever ``starttls`` says (GHSA-3588-2mr7-vrcm)."""
for starttls in (None, True, False):
with self.subTest(starttls=starttls):
calls = self._send(
advertise_starttls=False,
require_encryption=True,
starttls=starttls,
username="dmarc@example.com",
password=self.PASSWORD,
)
names = self._names(calls)
self.assertEqual(calls[0], ("SMTP_SSL", "smtp.example.com", 587))
self.assertNotIn("SMTP", names)
self.assertNotIn("starttls", names)
self.assertLess(names.index("login"), names.index("sendmail"))
def testNoCredentialsWithoutStartTLSSendsWithWarning(self):
"""Without credentials there is nothing to steal at login, so a
server without STARTTLS still gets the message in plain text, with
mailsuite's warning (GHSA-3588-2mr7-vrcm scope)."""
with self.assertLogs("mailsuite.smtp", level="WARNING") as logs:
calls = self._send(advertise_starttls=False)
names = self._names(calls)
self.assertNotIn("starttls", names)
self.assertNotIn("login", names)
self.assertEqual(names[-1], "sendmail")
self.assertTrue(
any("does not support STARTTLS" in line for line in logs.output),
logs.output,
)
def testStartTLSFalseAuthenticatesInPlainText(self):
"""``starttls=False`` is the explicit opt-out from
GHSA-3588-2mr7-vrcm's fix: credentials are sent without STARTTLS,
with a warning."""
with self.assertLogs("mailsuite.smtp", level="WARNING") as logs:
calls = self._send(
advertise_starttls=False,
starttls=False,
username="dmarc@example.com",
password=self.PASSWORD,
)
names = self._names(calls)
self.assertNotIn("starttls", names)
self.assertEqual(
calls[names.index("login")],
("login", "dmarc@example.com", self.PASSWORD),
)
self.assertLess(names.index("login"), names.index("sendmail"))
self.assertTrue(any("unencrypted" in line for line in logs.output), logs.output)
def testStartTLSFalseNeverUpgradesEvenWhenAdvertised(self):
"""``starttls=False`` never upgrades, even when the server
advertises STARTTLS (GHSA-3588-2mr7-vrcm opt-out semantics)."""
calls = self._send(
advertise_starttls=True,
starttls=False,
username="dmarc@example.com",
password=self.PASSWORD,
)
names = self._names(calls)
self.assertNotIn("starttls", names)
self.assertLess(names.index("login"), names.index("sendmail"))
def testStartTLSTrueRefusesWithoutAdvertisement(self):
"""``starttls=True`` requires the upgrade: no STARTTLS in the EHLO
reply raises before anything is sent, even without credentials
(GHSA-3588-2mr7-vrcm)."""
from mailsuite.smtp import SMTPError
with self.assertRaises(SMTPError):
self._send(advertise_starttls=False, starttls=True)
names = self._names(self.calls)
self.assertNotIn("sendmail", names)
self.assertNotIn("login", names)
class TestEmailResultsViaMsGraph(unittest.TestCase):
"""email_results_via_msgraph() shares its
subject/message/attachment-building logic with email_results() via the