test_email_required_headers.py 5.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143
  1. """Outgoing mail carries Date and Message-ID, and a readable From (#3376).
  2. Neither smtplib nor the MIME classes add Date or Message-ID, so strict
  3. receivers (amavis's bad_header policy) quarantined every Bambuddy mail after
  4. the SMTP server had accepted it. Both senders are covered: send_email (SMTP
  5. test, welcome, password reset, MFA) and the notification provider.
  6. """
  7. from email import message_from_string, policy
  8. from email.utils import parsedate_to_datetime
  9. from unittest.mock import patch
  10. import pytest
  11. from backend.app.schemas.auth import SMTPSettings
  12. from backend.app.services.email_service import add_required_headers, send_email
  13. from backend.app.services.notification_service import NotificationService
  14. JPEG = b"\xff\xd8\xff\xe0jpeg"
  15. PHOTO_URL = "https://printer.local/api/v1/archives/42/photos/finish.jpg"
  16. def _assert_required_headers(msg, domain: str) -> None:
  17. assert parsedate_to_datetime(msg["Date"]).tzinfo is not None
  18. message_id = msg["Message-ID"]
  19. assert message_id.startswith("<")
  20. assert message_id.endswith(f"@{domain}>")
  21. def _fake_smtp(captured: list):
  22. class FakeSMTP:
  23. def __init__(self, host, port, timeout=None):
  24. pass
  25. def __enter__(self):
  26. return self
  27. def __exit__(self, *exc):
  28. return False
  29. def starttls(self):
  30. pass
  31. def login(self, u, p):
  32. pass
  33. def send_message(self, msg):
  34. # As sent: serialise and re-parse, so header encoding counts.
  35. captured.append(message_from_string(msg.as_string()))
  36. def sendmail(self, frm, to, body):
  37. captured.append(message_from_string(body))
  38. def quit(self):
  39. pass
  40. return FakeSMTP
  41. def _smtp_settings(**kw) -> SMTPSettings:
  42. base = {
  43. "smtp_host": "smtp.example.com",
  44. "smtp_port": 587,
  45. "smtp_security": "starttls",
  46. "smtp_auth_enabled": False,
  47. "smtp_from_email": "farm@mail.example.org",
  48. "smtp_from_name": "Bambuddy",
  49. }
  50. base.update(kw)
  51. return SMTPSettings(**base)
  52. @pytest.mark.parametrize("image", [None, JPEG], ids=["alternative", "related"])
  53. def test_send_email_sets_date_and_message_id(image):
  54. captured: list = []
  55. with patch("backend.app.services.email_service.smtplib.SMTP", _fake_smtp(captured)):
  56. send_email(_smtp_settings(), "to@example.com", "Test", "body", "<p>body</p>", image_data=image)
  57. (msg,) = captured
  58. _assert_required_headers(msg, "mail.example.org")
  59. @pytest.mark.parametrize("image", [None, JPEG], ids=["alternative", "related"])
  60. def test_send_email_from_keeps_the_address_readable(image):
  61. """A non-ASCII or comma'd sender name is encoded on its own; the address
  62. stays plain, so receivers still find a valid From."""
  63. captured: list = []
  64. settings = _smtp_settings(smtp_from_name="Bämbuddy Farm, Inc")
  65. with patch("backend.app.services.email_service.smtplib.SMTP", _fake_smtp(captured)):
  66. send_email(settings, "to@example.com", "Test", "body", "<p>body</p>", image_data=image)
  67. (msg,) = captured
  68. assert "<farm@mail.example.org>" in msg["From"]
  69. # Decoded the way a mail client does.
  70. (sender,) = message_from_string(msg.as_string(), policy=policy.default)["From"].addresses
  71. assert (sender.display_name, sender.addr_spec) == ("Bämbuddy Farm, Inc", "farm@mail.example.org")
  72. @pytest.mark.asyncio
  73. @pytest.mark.parametrize("inline_photo", [False, True], ids=["text", "related"])
  74. async def test_notification_email_sets_date_and_message_id(inline_photo):
  75. config = {
  76. "smtp_server": "smtp.example.com",
  77. "smtp_port": "587",
  78. "from_email": "printer@farm.example.net",
  79. "to_email": "alice@example.com",
  80. "security": "starttls",
  81. "auth_enabled": "false",
  82. }
  83. body = f"Done.\n\nSnapshot: {PHOTO_URL}" if inline_photo else "Done."
  84. captured: list = []
  85. with patch("backend.app.services.notification_service.smtplib.SMTP", _fake_smtp(captured)):
  86. ok, _ = await NotificationService()._send_email(
  87. config, "Print Finished", body, image_data=JPEG, finish_photo_url=PHOTO_URL
  88. )
  89. assert ok is True
  90. (msg,) = captured
  91. assert msg.get_content_type() == ("multipart/related" if inline_photo else "multipart/mixed")
  92. _assert_required_headers(msg, "farm.example.net")
  93. @pytest.mark.parametrize("from_email", ["", "no-at-sign", "user@"])
  94. def test_message_id_without_a_sender_domain_uses_a_fixed_one(from_email):
  95. """Never make_msgid()'s hostname lookup: it can block on DNS and puts
  96. the container's name in the header."""
  97. from email.mime.multipart import MIMEMultipart
  98. msg = MIMEMultipart()
  99. with patch("socket.getfqdn", side_effect=AssertionError("hostname lookup")):
  100. add_required_headers(msg, from_email)
  101. _assert_required_headers(msg, "bambuddy.local")
  102. def test_existing_headers_are_kept():
  103. from email.mime.multipart import MIMEMultipart
  104. msg = MIMEMultipart()
  105. msg["Date"] = "Sat, 24 Oct 2026 19:40:19 +0000"
  106. msg["Message-ID"] = "<fixed@example.com>"
  107. add_required_headers(msg, "a@b.example")
  108. assert msg.get_all("Date") == ["Sat, 24 Oct 2026 19:40:19 +0000"]
  109. assert msg.get_all("Message-ID") == ["<fixed@example.com>"]