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 # The lookbehind keeps us inside a tag: without it a url ending in "src=" (base64 tokens are "=" padded, 

48 # so this happens) matches from within its own href across into the following attribute. 

49 html = re.sub(r'(?<=\s)src="([^":]+)"', repl=process_relative_src_match, string=html) 

50 return html, related_parts 

51 

52 

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

54 msg = EmailMessage() 

55 msg["Subject"] = payload.subject 

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

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

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

59 

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

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

62 

63 if payload.source_data: 

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

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

66 

67 msg.set_content(payload.plain) 

68 

69 if payload.html: 

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

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

72 

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

74 for related_part in payload.html_related_parts: 

75 _add_email_part(html_part, related_part, related=True) 

76 

77 if payload.attachments: 

78 for attachment in payload.attachments: 

79 _add_email_part(msg, attachment, related=False) 

80 

81 return msg 

82 

83 

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

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

86 data = part.data 

87 if not data and part.data_file_path: 

88 data_path = Path(part.data_file_path) 

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

90 data_path = email_related_part_data_path_base / data_path 

91 data = data_path.read_bytes() 

92 

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

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

95 if related: 

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

97 else: 

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

99 

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

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

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

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

104 if part.content_id: 

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

106 

107 return mime_part 

108 

109 

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

111 """ 

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

113 

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

115 """ 

116 message_id = random_hex() 

117 msg = email_proto_to_message(payload, message_id) 

118 

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

120 server.ehlo() 

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

122 server.starttls() 

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

124 server.ehlo() 

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

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

127 

128 return Email( 

129 message_id=message_id, 

130 sender_name=payload.sender_name, 

131 sender_email=payload.sender_email, 

132 recipient=payload.recipient, 

133 subject=payload.subject, 

134 plain=payload.plain, 

135 html=payload.html, 

136 list_unsubscribe_header=payload.list_unsubscribe_header, 

137 source_data=payload.source_data, 

138 ) 

139 

140 

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

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

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

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

145 

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

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

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

149 if isinstance(header_index, int): 

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

151 else: 

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

153 else: 

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

155 part.replace_header(name, value)