Coverage for app/backend/src/couchers/email/smtp.py: 89%

1import re 

2import smtplib 

3from email.headerregistry import Address 

4from email.message import EmailMessage, MIMEPart 

5from email.utils import make_msgid 

6from pathlib import Path 

7from typing import cast 

8 

9import couchers 

10from couchers.config import config 

11from couchers.crypto import EMAIL_SOURCE_DATA_KEY_NAME, random_hex, simple_hash_signature 

12from couchers.models import Email 

13from couchers.proto.internal import jobs_pb2 

14 

15# Base directory for relative EmailPart.data_file_path 

16email_related_part_data_path_base = Path(couchers.__file__).parents[3] # /app/backend 

17 

18 

19def embed_html_relative_images( 

20 html: str, *, base_dir: Path, content_id_domain: str 

21) -> tuple[str, list[jobs_pb2.EmailPart]]: 

22 """Modifies HTML markup's image references such that they can be embedded in multipart/related MIME parts.""" 

23 related_parts: list[jobs_pb2.EmailPart] = [] 

24 

25 def process_relative_src_match(match: re.Match[str]) -> str: 

26 """Replaces a src="" attribute with a content id reference.""" 

27 image_path = base_dir / str(match.group(1)) 

28 if not image_path.exists(): 28 ↛ 29line 28 didn't jump to line 29 because the condition on line 28 was never true

29 raise FileExistsError(f"HTML references missing relative image: {image_path}") 

30 

31 root_relative_path = image_path.relative_to(email_related_part_data_path_base) 

32 mime_type = f"image/{image_path.suffix.removeprefix('.')}" 

33 filename = image_path.name 

34 bracketed_content_id = make_msgid(domain=content_id_domain) 

35 content_id = bracketed_content_id[1:-1] 

36 related_parts.append( 

37 jobs_pb2.EmailPart( 

38 data_file_path=str(root_relative_path), 

39 content_type=f'{mime_type}; name="{filename}"', 

40 content_disposition=f'inline; filename="{filename}"', 

41 content_id=bracketed_content_id, 

42 ) 

43 ) 

44 

45 return f'src="cid:{content_id}"' 

46 

47 html = re.sub(r'src="([^":]+)"', repl=process_relative_src_match, string=html) 

48 return html, related_parts 

49 

50 

51def email_proto_to_message(payload: jobs_pb2.SendEmailPayload, couchers_id: str) -> EmailMessage: 

52 msg = EmailMessage() 

53 msg["Subject"] = payload.subject 

54 msg["From"] = Address(payload.sender_name, addr_spec=payload.sender_email) 

55 msg["To"] = Address(addr_spec=payload.recipient) 

56 msg["X-Couchers-ID"] = couchers_id 

57 

58 if payload.list_unsubscribe_header: 58 ↛ 59line 58 didn't jump to line 59 because the condition on line 58 was never true

59 msg["List-Unsubscribe"] = payload.list_unsubscribe_header 

60 

61 if payload.source_data: 

62 msg["X-Couchers-Source-Data"] = payload.source_data 

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

64 

65 msg.set_content(payload.plain) 

66 

67 if payload.html: 

68 msg.add_alternative(payload.html, subtype="html") 

69 html_part = cast(list[MIMEPart], msg.get_payload())[-1] 

70 

71 if payload.html_related_parts: 71 ↛ 75line 71 didn't jump to line 75 because the condition on line 71 was always true

72 for related_part in payload.html_related_parts: 

73 _add_email_part(html_part, related_part, related=True) 

74 

75 if payload.attachments: 

76 for attachment in payload.attachments: 

77 _add_email_part(msg, attachment, related=False) 

78 

79 return msg 

80 

81 

82def _add_email_part(msg: MIMEPart, part: jobs_pb2.EmailPart, *, related: bool) -> MIMEPart: 

83 # The data is either part of the payload or must be loaded from a file 

84 data = part.data 

85 if not data and part.data_file_path: 

86 data_path = Path(part.data_file_path) 

87 if not data_path.is_absolute(): 87 ↛ 89line 87 didn't jump to line 89 because the condition on line 87 was always true

88 data_path = email_related_part_data_path_base / data_path 

89 data = data_path.read_bytes() 

90 

91 # Create with generic Content-Type/Content-Disposition headers, 

92 # then overwrite them with the headers specified by the caller. 

93 if related: 

94 msg.add_related(data, maintype="application", subtype="octet-stream", disposition="inline") 

95 else: 

96 msg.add_attachment(data, maintype="application", subtype="octet-stream", disposition="attachment") 

97 

98 mime_part = cast(list[MIMEPart], msg.get_payload())[-1] 

99 _replace_header_verbatim(mime_part, "Content-Type", part.content_type) 

100 if part.content_disposition: 100 ↛ 102line 100 didn't jump to line 102 because the condition on line 100 was always true

101 _replace_header_verbatim(mime_part, "Content-Disposition", part.content_disposition) 

102 if part.content_id: 

103 _replace_header_verbatim(mime_part, "Content-ID", part.content_id) 

104 

105 return mime_part 

106 

107 

108def send_smtp_email(payload: jobs_pb2.SendEmailPayload) -> Email: 

109 """ 

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

111 

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

113 """ 

114 message_id = random_hex() 

115 msg = email_proto_to_message(payload, message_id) 

116 

117 with smtplib.SMTP(config.SMTP_HOST, config.SMTP_PORT) as server: 

118 server.ehlo() 

119 if not config.DEV: 119 ↛ 120line 119 didn't jump to line 120 because the condition on line 119 was never true

120 server.starttls() 

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

122 server.ehlo() 

123 server.login(config.SMTP_USERNAME, config.SMTP_PASSWORD) 

124 server.sendmail(payload.sender_email, payload.recipient, msg.as_string()) 

125 

126 return Email( 

127 id=message_id, 

128 sender_name=payload.sender_name, 

129 sender_email=payload.sender_email, 

130 recipient=payload.recipient, 

131 subject=payload.subject, 

132 plain=payload.plain, 

133 html=payload.html, 

134 list_unsubscribe_header=payload.list_unsubscribe_header, 

135 source_data=payload.source_data, 

136 ) 

137 

138 

139def _replace_header_verbatim(part: MIMEPart, name: str, value: str) -> None: 

140 # MIMEPart.replace_header will parse the value and reformat it, 

141 # resulting in additional quoting for an .ics "method=PUBLISH" parameter, 

142 # which are not as backwards compatible with older email clients. 

143 

144 if hasattr(part, "_headers"): 144 ↛ 153line 144 didn't jump to line 153 because the condition on line 144 was always true

145 # Replace the header in the internal data structure to avoid reformatting. 

146 header_index = next((i for i, val in enumerate(part._headers) if val[0] == name), None) 

147 if isinstance(header_index, int): 

148 part._headers[header_index] = (name, value) 

149 else: 

150 part._headers.append((name, value)) 

151 else: 

152 # Non-verbatim fallback, in case the internals change 

153 part.replace_header(name, value)