Coverage for src/couchers/email/smtp.py: 90%

49 statements  

« prev     ^ index     » next       coverage.py v7.11.0, created at 2025-12-06 23:17 +0000

1import smtplib 

2from email.headerregistry import Address 

3from email.message import EmailMessage, MIMEPart 

4from email.utils import make_msgid 

5from pathlib import Path 

6from typing import cast 

7 

8from couchers.config import config 

9from couchers.crypto import EMAIL_SOURCE_DATA_KEY_NAME, random_hex, simple_hash_signature 

10from couchers.models import Email 

11 

12template_base = Path(Path(__file__).parent / ".." / ".." / ".." / "templates" / "v2") 

13 

14 

15def make_cid(sender_email: str) -> tuple[str, str]: 

16 cid = make_msgid(domain=Address(addr_spec=sender_email).domain) 

17 without_tag = cid[1:-1] 

18 return cid, without_tag 

19 

20 

21def send_smtp_email( 

22 sender_name: str, 

23 sender_email: str, 

24 recipient: str, 

25 subject: str, 

26 plain: str, 

27 html: str | None, 

28 list_unsubscribe_header: str | None, 

29 source_data: str | None, 

30) -> Email: 

31 """ 

32 Sends out the email through SMTP, settings from config. 

33 

34 Returns a models.Email object that can be straight away added to the database. 

35 """ 

36 message_id = random_hex() 

37 msg = EmailMessage() 

38 msg["Subject"] = subject 

39 msg["From"] = Address(sender_name, addr_spec=sender_email) 

40 msg["To"] = Address(addr_spec=recipient) 

41 msg["X-Couchers-ID"] = message_id 

42 

43 if list_unsubscribe_header: 

44 msg["List-Unsubscribe"] = list_unsubscribe_header 

45 

46 if source_data: 

47 msg["X-Couchers-Source-Data"] = source_data 

48 msg["X-Couchers-Source-Sig"] = simple_hash_signature(source_data, EMAIL_SOURCE_DATA_KEY_NAME) 

49 

50 msg.set_content(plain) 

51 

52 if html: 

53 # for any png files in attachment_imgs/, goes through and replaces instances of the filename with attachment 

54 used_attachments = [] 

55 for attachment in (template_base / "attachment_imgs").glob("*.png"): 

56 attachment_html_path = str(attachment.relative_to(template_base)) 

57 if attachment_html_path not in html: 

58 continue 

59 # it's used in this template, so attach and replace it 

60 data = attachment.read_bytes() 

61 cid, wcid = make_cid(sender_email) 

62 html = html.replace(attachment_html_path, f"cid:{wcid}") 

63 used_attachments.append((cid, "image", "png", data)) 

64 

65 msg.add_alternative(html, subtype="html") 

66 

67 for cid, mime_type, mime_subtype, data in used_attachments: 

68 payloads = cast(list[MIMEPart], msg.get_payload()) 

69 payloads[1].add_related(data, mime_type, mime_subtype, cid=cid) 

70 

71 with smtplib.SMTP(config["SMTP_HOST"], config["SMTP_PORT"]) as server: 

72 server.ehlo() 

73 if not config["DEV"]: 

74 server.starttls() 

75 # stmplib docs recommend calling ehlo() before and after starttls() 

76 server.ehlo() 

77 server.login(config["SMTP_USERNAME"], config["SMTP_PASSWORD"]) 

78 server.sendmail(sender_email, recipient, msg.as_string()) 

79 

80 return Email( 

81 id=message_id, 

82 sender_name=sender_name, 

83 sender_email=sender_email, 

84 recipient=recipient, 

85 subject=subject, 

86 plain=plain, 

87 html=html, 

88 list_unsubscribe_header=list_unsubscribe_header, 

89 source_data=source_data, 

90 )