1import mimetypes
2from email import charset as Charset
3from email import encoders as Encoders
4from email import generator, message_from_string
5from email.errors import HeaderParseError
6from email.header import Header
7from email.headerregistry import Address
8from email.headerregistry import (
9 parser as headerregistry_parser, # ty: ignore[unresolved-import]
10)
11from email.message import Message
12from email.mime.base import MIMEBase
13from email.mime.message import MIMEMessage
14from email.mime.multipart import MIMEMultipart
15from email.mime.text import MIMEText
16from email.utils import formataddr, formatdate, getaddresses, make_msgid
17from io import BytesIO, StringIO
18from pathlib import Path
19from typing import TYPE_CHECKING, Any
20
21from plain.runtime import settings
22from plain.templates import Template, TemplateFileMissing
23from plain.utils.encoding import force_str, punycode
24from plain.utils.html import strip_tags
25
26from .utils import _DNS_NAME
27
28if TYPE_CHECKING:
29 from os import PathLike
30
31 from .backends.base import BaseEmailBackend
32
33# Don't BASE64-encode UTF-8 messages so that we avoid unwanted attention from
34# some spam filters.
35_utf8_charset = Charset.Charset("utf-8")
36_utf8_charset.body_encoding = None # ty: ignore[invalid-assignment] # Python defaults to BASE64
37_utf8_charset_qp = Charset.Charset("utf-8")
38_utf8_charset_qp.body_encoding = Charset.QP
39
40# Default MIME type to use on attachments (if it is not explicitly given
41# and cannot be guessed).
42_DEFAULT_ATTACHMENT_MIME_TYPE = "application/octet-stream"
43
44_RFC5322_EMAIL_LINE_LENGTH_LIMIT = 998
45
46
47class BadHeaderError(ValueError):
48 pass
49
50
51# Header names that contain structured address data (RFC 5322).
52_ADDRESS_HEADERS = {
53 "from",
54 "sender",
55 "reply-to",
56 "to",
57 "cc",
58 "bcc",
59 "resent-from",
60 "resent-sender",
61 "resent-to",
62 "resent-cc",
63 "resent-bcc",
64}
65
66
67def _forbid_multi_line_headers(
68 name: str, val: str, encoding: str | None
69) -> tuple[str, str]:
70 """Forbid multi-line headers to prevent header injection."""
71 encoding = encoding or "utf-8"
72 val = str(val) # val may be lazy
73 if "\n" in val or "\r" in val:
74 raise BadHeaderError(
75 f"Header values can't contain newlines (got {val!r} for header {name!r})"
76 )
77 try:
78 val.encode("ascii")
79 except UnicodeEncodeError:
80 if name.lower() in _ADDRESS_HEADERS:
81 val = ", ".join(
82 _sanitize_address(addr, encoding) for addr in getaddresses((val,))
83 )
84 else:
85 val = Header(val, encoding).encode()
86 else:
87 if name.lower() == "subject":
88 val = Header(val).encode()
89 return name, val
90
91
92def _sanitize_address(addr: str | tuple[str, str], encoding: str) -> str:
93 """
94 Format a pair of (name, address) or an email address string.
95 """
96 address = None
97 if not isinstance(addr, tuple):
98 addr = force_str(addr)
99 try:
100 token, rest = headerregistry_parser.get_mailbox(addr)
101 except HeaderParseError, ValueError, IndexError:
102 raise ValueError(f'Invalid address "{addr}"')
103 else:
104 if rest:
105 # The entire email address must be parsed.
106 raise ValueError(
107 f'Invalid address; only {token} could be parsed from "{addr}"'
108 )
109 nm = token.display_name or ""
110 localpart = token.local_part
111 domain = token.domain or ""
112 else:
113 nm, address = addr
114 localpart, domain = address.rsplit("@", 1)
115
116 address_parts = nm + localpart + domain
117 if "\n" in address_parts or "\r" in address_parts:
118 raise ValueError("Invalid address; address parts cannot contain newlines.")
119
120 # Avoid UTF-8 encode, if it's possible.
121 try:
122 nm.encode("ascii")
123 nm = Header(nm).encode()
124 except UnicodeEncodeError:
125 nm = Header(nm, encoding).encode()
126 try:
127 localpart.encode("ascii")
128 except UnicodeEncodeError:
129 localpart = Header(localpart, encoding).encode()
130 domain = punycode(domain)
131
132 parsed_address = Address(username=localpart, domain=domain)
133 return formataddr((nm, parsed_address.addr_spec))
134
135
136class MIMEMixin:
137 def as_string(self, unixfrom: bool = False, linesep: str = "\n") -> str:
138 """Return the entire formatted message as a string.
139 Optional `unixfrom' when True, means include the Unix From_ envelope
140 header.
141
142 This overrides the default as_string() implementation to not mangle
143 lines that begin with 'From '. See bug #13433 for details.
144 """
145 fp = StringIO()
146 g = generator.Generator(fp, mangle_from_=False)
147 g.flatten(self, unixfrom=unixfrom, linesep=linesep)
148 return fp.getvalue()
149
150 def as_bytes(self, unixfrom: bool = False, linesep: str = "\n") -> bytes:
151 """Return the entire formatted message as bytes.
152 Optional `unixfrom' when True, means include the Unix From_ envelope
153 header.
154
155 This overrides the default as_bytes() implementation to not mangle
156 lines that begin with 'From '. See bug #13433 for details.
157 """
158 fp = BytesIO()
159 g = generator.BytesGenerator(fp, mangle_from_=False)
160 g.flatten(self, unixfrom=unixfrom, linesep=linesep)
161 return fp.getvalue()
162
163
164class SafeMIMEMessage(MIMEMixin, MIMEMessage): # ty: ignore[invalid-method-override]
165 def __setitem__(self, name: str, val: str) -> None:
166 # message/rfc822 attachments must be ASCII
167 name, val = _forbid_multi_line_headers(name, val, "ascii")
168 MIMEMessage.__setitem__(self, name, val)
169
170
171class SafeMIMEText(MIMEMixin, MIMEText): # ty: ignore[invalid-method-override]
172 def __init__(
173 self, _text: str, _subtype: str = "plain", _charset: str | None = None
174 ) -> None:
175 self.encoding = _charset
176 MIMEText.__init__(self, _text, _subtype=_subtype, _charset=_charset)
177
178 def __setitem__(self, name: str, val: str) -> None:
179 name, val = _forbid_multi_line_headers(name, val, self.encoding)
180 MIMEText.__setitem__(self, name, val)
181
182 def set_payload( # ty: ignore[invalid-method-override]
183 self, payload: str, charset: str | Charset.Charset | None = None
184 ) -> None:
185 if charset == "utf-8" and not isinstance(charset, Charset.Charset):
186 has_long_lines = any(
187 len(line.encode()) > _RFC5322_EMAIL_LINE_LENGTH_LIMIT
188 for line in payload.splitlines()
189 )
190 # Quoted-Printable encoding has the side effect of shortening long
191 # lines, if any (#22561).
192 charset = _utf8_charset_qp if has_long_lines else _utf8_charset
193 MIMEText.set_payload(self, payload, charset=charset)
194
195
196class SafeMIMEMultipart(MIMEMixin, MIMEMultipart): # ty: ignore[invalid-method-override]
197 def __init__(
198 self,
199 _subtype: str = "mixed",
200 boundary: str | None = None,
201 _subparts: list[Message] | None = None,
202 encoding: str | None = None,
203 **_params: Any,
204 ) -> None:
205 self.encoding = encoding
206 MIMEMultipart.__init__(self, _subtype, boundary, _subparts, **_params)
207
208 def __setitem__(self, name: str, val: str) -> None:
209 name, val = _forbid_multi_line_headers(name, val, self.encoding)
210 MIMEMultipart.__setitem__(self, name, val)
211
212
213class EmailMessage:
214 """A container for email information."""
215
216 content_subtype = "plain"
217 mixed_subtype = "mixed"
218 encoding: str | None = None # None => use settings default
219
220 def __init__(
221 self,
222 subject: str = "",
223 body: str = "",
224 from_email: str | None = None,
225 to: list[str] | tuple[str, ...] | None = None,
226 bcc: list[str] | tuple[str, ...] | None = None,
227 connection: BaseEmailBackend | None = None,
228 attachments: list[MIMEBase | tuple[str, str, str]] | None = None,
229 headers: dict[str, str] | None = None,
230 cc: list[str] | tuple[str, ...] | None = None,
231 reply_to: list[str] | tuple[str, ...] | None = None,
232 ) -> None:
233 """
234 Initialize a single email message (which can be sent to multiple
235 recipients).
236 """
237 if to:
238 if isinstance(to, str):
239 raise TypeError('"to" argument must be a list or tuple')
240 self.to = list(to)
241 else:
242 self.to = []
243 if cc:
244 if isinstance(cc, str):
245 raise TypeError('"cc" argument must be a list or tuple')
246 self.cc = list(cc)
247 else:
248 self.cc = []
249 if bcc:
250 if isinstance(bcc, str):
251 raise TypeError('"bcc" argument must be a list or tuple')
252 self.bcc = list(bcc)
253 else:
254 self.bcc = []
255 if reply_to:
256 if isinstance(reply_to, str):
257 raise TypeError('"reply_to" argument must be a list or tuple')
258 self.reply_to = list(reply_to)
259 else:
260 self.reply_to = settings.EMAIL_DEFAULT_REPLY_TO or []
261 self.from_email = from_email or settings.EMAIL_DEFAULT_FROM
262 self.subject = subject
263 self.body = body or ""
264 self.attachments = []
265 if attachments:
266 for attachment in attachments:
267 if isinstance(attachment, MIMEBase):
268 self.attach(attachment)
269 else:
270 self.attach(*attachment)
271 self.extra_headers = headers or {}
272 self.connection = connection
273
274 def get_connection(self) -> BaseEmailBackend:
275 from . import get_connection
276
277 if not self.connection:
278 self.connection = get_connection()
279 return self.connection
280
281 def message(self) -> SafeMIMEText | SafeMIMEMultipart:
282 encoding = self.encoding or "utf-8"
283 msg = SafeMIMEText(self.body, self.content_subtype, encoding)
284 msg = self._create_message(msg)
285 msg["Subject"] = self.subject
286 msg["From"] = self.extra_headers.get("From", self.from_email)
287 self._set_list_header_if_not_empty(msg, "To", self.to)
288 self._set_list_header_if_not_empty(msg, "Cc", self.cc)
289 self._set_list_header_if_not_empty(msg, "Reply-To", self.reply_to)
290
291 # Email header names are case-insensitive (RFC 2045), so we have to
292 # accommodate that when doing comparisons.
293 header_names = [key.lower() for key in self.extra_headers]
294 if "date" not in header_names:
295 # formatdate() uses stdlib methods to format the date, which use
296 # the stdlib/OS concept of a timezone, however, Plain sets the
297 # TZ environment variable based on the TIME_ZONE setting which
298 # will get picked up by formatdate().
299 msg["Date"] = formatdate(localtime=settings.EMAIL_USE_LOCALTIME)
300 if "message-id" not in header_names:
301 # Use cached _DNS_NAME for performance
302 msg["Message-ID"] = make_msgid(domain=str(_DNS_NAME))
303 for name, value in self.extra_headers.items():
304 if name.lower() != "from": # From is already handled
305 msg[name] = value
306 return msg
307
308 def recipients(self) -> list[str]:
309 """
310 Return a list of all recipients of the email (includes direct
311 addressees as well as Cc and Bcc entries).
312 """
313 return [email for email in (self.to + self.cc + self.bcc) if email]
314
315 def send(self) -> int:
316 """Send the email message."""
317 if not self.recipients():
318 # Don't bother creating the network connection if there's nobody to
319 # send to.
320 return 0
321 return self.get_connection().send_messages([self])
322
323 def attach(
324 self,
325 filename: MIMEBase | str | None = None,
326 content: str | bytes | None = None,
327 mimetype: str | None = None,
328 ) -> None:
329 """
330 Attach a file with the given filename and content. The filename can
331 be omitted and the mimetype is guessed, if not provided.
332
333 If the first parameter is a MIMEBase subclass, insert it directly
334 into the resulting message attachments.
335
336 For a text/* mimetype (guessed or specified), when a bytes object is
337 specified as content, decode it as UTF-8. If that fails, set the
338 mimetype to _DEFAULT_ATTACHMENT_MIME_TYPE and don't decode the content.
339 """
340 if isinstance(filename, MIMEBase):
341 if content is not None or mimetype is not None:
342 raise ValueError(
343 "content and mimetype must not be given when a MIMEBase "
344 "instance is provided."
345 )
346 self.attachments.append(filename)
347 elif content is None:
348 raise ValueError("content must be provided.")
349 else:
350 if filename is not None and mimetype is None:
351 mimetype = mimetypes.guess_type(filename)[0]
352 if mimetype is None:
353 mimetype = _DEFAULT_ATTACHMENT_MIME_TYPE
354 basetype, _subtype = mimetype.split("/", 1)
355
356 if basetype == "text" and isinstance(content, bytes):
357 try:
358 content = content.decode()
359 except UnicodeDecodeError:
360 # If mimetype suggests the file is text but it's
361 # actually binary, read() raises a UnicodeDecodeError.
362 mimetype = _DEFAULT_ATTACHMENT_MIME_TYPE
363
364 self.attachments.append((filename, content, mimetype))
365
366 def attach_file(
367 self, path: str | PathLike[str], mimetype: str | None = None
368 ) -> None:
369 """
370 Attach a file from the filesystem.
371
372 Set the mimetype to _DEFAULT_ATTACHMENT_MIME_TYPE if it isn't specified
373 and cannot be guessed.
374
375 For a text/* mimetype (guessed or specified), decode the file's content
376 as UTF-8. If that fails, set the mimetype to
377 _DEFAULT_ATTACHMENT_MIME_TYPE and don't decode the content.
378 """
379 path = Path(path)
380 with path.open("rb") as file:
381 content = file.read()
382 self.attach(path.name, content, mimetype)
383
384 def _create_message(self, msg: SafeMIMEText) -> SafeMIMEText | SafeMIMEMultipart:
385 return self._create_attachments(msg)
386
387 def _create_attachments(
388 self, msg: SafeMIMEText | SafeMIMEMultipart
389 ) -> SafeMIMEText | SafeMIMEMultipart:
390 if self.attachments:
391 encoding = self.encoding or "utf-8"
392 body_msg = msg
393 msg = SafeMIMEMultipart(_subtype=self.mixed_subtype, encoding=encoding)
394 if self.body or body_msg.is_multipart():
395 msg.attach(body_msg)
396 for attachment in self.attachments:
397 if isinstance(attachment, MIMEBase):
398 msg.attach(attachment)
399 else:
400 msg.attach(self._create_attachment(*attachment))
401 return msg
402
403 def _create_mime_attachment(
404 self, content: str | bytes | EmailMessage | Message, mimetype: str
405 ) -> SafeMIMEText | SafeMIMEMessage | MIMEBase:
406 """
407 Convert the content, mimetype pair into a MIME attachment object.
408
409 If the mimetype is message/rfc822, content may be an
410 email.Message or EmailMessage object, as well as a str.
411 """
412 basetype, subtype = mimetype.split("/", 1)
413 if basetype == "text":
414 encoding = self.encoding or "utf-8"
415 if not isinstance(content, str):
416 content = force_str(content)
417 attachment = SafeMIMEText(content, subtype, encoding)
418 elif basetype == "message" and subtype == "rfc822":
419 # Bug #18967: Per RFC 2046 Section 5.2.1, message/rfc822
420 # attachments must not be base64 encoded.
421 if isinstance(content, EmailMessage):
422 # convert content into an email.Message first
423 content = content.message()
424 elif not isinstance(content, Message):
425 # For compatibility with existing code, parse the message
426 # into an email.Message object if it is not one already.
427 content = message_from_string(force_str(content))
428
429 attachment = SafeMIMEMessage(content, subtype)
430 else:
431 # Encode non-text attachments with base64.
432 attachment = MIMEBase(basetype, subtype)
433 assert isinstance(content, bytes), "Non-text attachments must be bytes"
434 attachment.set_payload(content)
435 Encoders.encode_base64(attachment)
436 return attachment
437
438 def _create_attachment(
439 self, filename: str | None, content: str | bytes, mimetype: str
440 ) -> SafeMIMEText | SafeMIMEMessage | MIMEBase:
441 """
442 Convert the filename, content, mimetype triple into a MIME attachment
443 object.
444 """
445 attachment = self._create_mime_attachment(content, mimetype)
446 if filename:
447 try:
448 filename.encode("ascii")
449 encoded_filename: str | tuple[str, str, str] = filename
450 except UnicodeEncodeError:
451 encoded_filename = ("utf-8", "", filename)
452 attachment.add_header(
453 "Content-Disposition", "attachment", filename=encoded_filename
454 )
455 return attachment
456
457 def _set_list_header_if_not_empty(
458 self,
459 msg: SafeMIMEText | SafeMIMEMultipart,
460 header: str,
461 values: list[str],
462 ) -> None:
463 """
464 Set msg's header, either from self.extra_headers, if present, or from
465 the values argument.
466 """
467 if values:
468 try:
469 value = self.extra_headers[header]
470 except KeyError:
471 value = ", ".join(str(v) for v in values)
472 msg[header] = value
473
474
475class EmailMultiAlternatives(EmailMessage):
476 """
477 A version of EmailMessage that makes it easy to send multipart/alternative
478 messages. For example, including text and HTML versions of the text is
479 made easier.
480 """
481
482 alternative_subtype = "alternative"
483
484 def __init__(
485 self,
486 subject: str = "",
487 body: str = "",
488 from_email: str | None = None,
489 to: list[str] | tuple[str, ...] | None = None,
490 bcc: list[str] | tuple[str, ...] | None = None,
491 connection: BaseEmailBackend | None = None,
492 attachments: list[MIMEBase | tuple[str, str, str]] | None = None,
493 headers: dict[str, str] | None = None,
494 alternatives: list[tuple[str, str]] | None = None,
495 cc: list[str] | tuple[str, ...] | None = None,
496 reply_to: list[str] | tuple[str, ...] | None = None,
497 ) -> None:
498 """
499 Initialize a single email message (which can be sent to multiple
500 recipients).
501 """
502 super().__init__(
503 subject,
504 body,
505 from_email,
506 to,
507 bcc,
508 connection,
509 attachments,
510 headers,
511 cc,
512 reply_to,
513 )
514 self.alternatives = alternatives or []
515
516 def attach_alternative(self, content: str, mimetype: str) -> None:
517 """Attach an alternative content representation."""
518 if content is None or mimetype is None:
519 raise ValueError("Both content and mimetype must be provided.")
520 self.alternatives.append((content, mimetype))
521
522 def _create_message(self, msg: SafeMIMEText) -> SafeMIMEText | SafeMIMEMultipart:
523 return self._create_attachments(self._create_alternatives(msg))
524
525 def _create_alternatives(
526 self, msg: SafeMIMEText | SafeMIMEMultipart
527 ) -> SafeMIMEText | SafeMIMEMultipart:
528 encoding = self.encoding or "utf-8"
529 if self.alternatives:
530 body_msg = msg
531 msg = SafeMIMEMultipart(
532 _subtype=self.alternative_subtype, encoding=encoding
533 )
534 if self.body:
535 msg.attach(body_msg)
536 for alternative in self.alternatives:
537 msg.attach(self._create_mime_attachment(*alternative))
538 return msg
539
540
541class TemplateEmail(EmailMultiAlternatives):
542 def __init__(
543 self,
544 *,
545 template: str,
546 context: dict[str, Any] | None = None,
547 subject: str = "",
548 from_email: str | None = None,
549 to: list[str] | tuple[str, ...] | None = None,
550 bcc: list[str] | tuple[str, ...] | None = None,
551 connection: BaseEmailBackend | None = None,
552 attachments: list[MIMEBase | tuple[str, str, str]] | None = None,
553 headers: dict[str, str] | None = None,
554 alternatives: list[tuple[str, str]] | None = None,
555 cc: list[str] | tuple[str, ...] | None = None,
556 reply_to: list[str] | tuple[str, ...] | None = None,
557 ) -> None:
558 self.template = template
559 self.context = context or {}
560
561 # Run this once for all uses of the context
562 render_context = self.get_template_context()
563
564 self.body_html, body = self.render_content(render_context)
565
566 if not subject:
567 subject = self.render_subject(render_context)
568
569 super().__init__(
570 subject=subject,
571 body=body,
572 from_email=from_email,
573 to=to,
574 bcc=bcc,
575 connection=connection,
576 attachments=attachments,
577 headers=headers,
578 alternatives=alternatives,
579 cc=cc,
580 reply_to=reply_to,
581 )
582
583 self.attach_alternative(self.body_html, "text/html")
584
585 def get_template_context(self) -> dict[str, Any]:
586 """Subclasses can override this method to add context data."""
587 return self.context
588
589 def render_content(self, context: dict[str, Any]) -> tuple[str, str]:
590 html_content = self.render_html(context)
591
592 try:
593 plain_content = self.render_plain(context)
594 except TemplateFileMissing:
595 plain_content = strip_tags(html_content)
596
597 return html_content, plain_content
598
599 def render_plain(self, context: dict[str, Any]) -> str:
600 return Template(self.get_plain_template_name()).render(context)
601
602 def render_html(self, context: dict[str, Any]) -> str:
603 return Template(self.get_html_template_name()).render(context)
604
605 def render_subject(self, context: dict[str, Any]) -> str:
606 try:
607 subject = Template(self.get_subject_template_name()).render(context)
608 return subject.strip()
609 except TemplateFileMissing:
610 return ""
611
612 def get_plain_template_name(self) -> str:
613 return f"email/{self.template}.txt"
614
615 def get_html_template_name(self) -> str:
616 return f"email/{self.template}.html"
617
618 def get_subject_template_name(self) -> str:
619 return f"email/{self.template}.subject.txt"