435 lines
17 KiB
Python
435 lines
17 KiB
Python
""" Email client to forge and send emails """
|
|
|
|
import base64
|
|
import email.utils
|
|
import logging
|
|
import os
|
|
import smtplib
|
|
from email.encoders import encode_base64
|
|
from email.mime.base import MIMEBase
|
|
from email.mime.multipart import MIMEMultipart
|
|
from email.mime.text import MIMEText
|
|
|
|
import magic
|
|
from mako.template import Template as MakoTemplate
|
|
|
|
from mylib.config import (
|
|
BooleanOption,
|
|
ConfigurableObject,
|
|
IntegerOption,
|
|
PasswordOption,
|
|
StringOption,
|
|
)
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
def load_image_as_base64(path):
|
|
"""Load image file as base64"""
|
|
log.debug("Load image file '%s'", path)
|
|
with open(path, "rb") as file_desc:
|
|
data = file_desc.read()
|
|
return f"data:{magic.from_buffer(data, mime=True)};base64, {base64.b64encode(data).decode()}"
|
|
|
|
|
|
class EmailClient(
|
|
ConfigurableObject
|
|
): # pylint: disable=useless-object-inheritance,too-many-instance-attributes
|
|
"""
|
|
Email client
|
|
|
|
This class abstract all interactions with the SMTP server.
|
|
"""
|
|
|
|
_config_name = "email"
|
|
_config_comment = "Email"
|
|
_defaults = {
|
|
"smtp_host": "localhost",
|
|
"smtp_port": 25,
|
|
"smtp_ssl": False,
|
|
"smtp_tls": False,
|
|
"smtp_user": None,
|
|
"smtp_password": None,
|
|
"smtp_debug": False,
|
|
"sender_name": "No reply",
|
|
"sender_email": "noreply@localhost",
|
|
"encoding": "utf-8",
|
|
"catch_all_addr": None,
|
|
"just_try": False,
|
|
"templates_path": None,
|
|
}
|
|
|
|
templates = {}
|
|
|
|
def __init__(self, templates=None, initialize=False, **kwargs):
|
|
super().__init__(**kwargs)
|
|
|
|
assert templates is None or isinstance(templates, dict)
|
|
self.templates = templates if templates else {}
|
|
if initialize:
|
|
self.initialize()
|
|
|
|
# pylint: disable=arguments-differ,arguments-renamed
|
|
def configure(self, use_smtp=True, **kwargs):
|
|
"""Configure options on registered mylib.Config object"""
|
|
section = super().configure(
|
|
just_try_help=kwargs.pop("just_try_help", "Just-try mode: do not really send emails"),
|
|
**kwargs,
|
|
)
|
|
|
|
if use_smtp:
|
|
section.add_option(
|
|
StringOption,
|
|
"smtp_host",
|
|
default=self._defaults["smtp_host"],
|
|
comment="SMTP server hostname/IP address",
|
|
)
|
|
section.add_option(
|
|
IntegerOption,
|
|
"smtp_port",
|
|
default=self._defaults["smtp_port"],
|
|
comment="SMTP server port",
|
|
)
|
|
section.add_option(
|
|
BooleanOption,
|
|
"smtp_ssl",
|
|
default=self._defaults["smtp_ssl"],
|
|
comment="Use SSL on SMTP server connection",
|
|
)
|
|
section.add_option(
|
|
BooleanOption,
|
|
"smtp_tls",
|
|
default=self._defaults["smtp_tls"],
|
|
comment="Use TLS on SMTP server connection",
|
|
)
|
|
section.add_option(
|
|
StringOption,
|
|
"smtp_user",
|
|
default=self._defaults["smtp_user"],
|
|
comment="SMTP authentication username",
|
|
)
|
|
section.add_option(
|
|
PasswordOption,
|
|
"smtp_password",
|
|
default=self._defaults["smtp_password"],
|
|
comment='SMTP authentication password (set to "keyring" to use XDG keyring)',
|
|
username_option="smtp_user",
|
|
keyring_value="keyring",
|
|
)
|
|
section.add_option(
|
|
BooleanOption,
|
|
"smtp_debug",
|
|
default=self._defaults["smtp_debug"],
|
|
comment="Enable SMTP debugging",
|
|
)
|
|
|
|
section.add_option(
|
|
StringOption,
|
|
"sender_name",
|
|
default=self._defaults["sender_name"],
|
|
comment="Sender name",
|
|
)
|
|
section.add_option(
|
|
StringOption,
|
|
"sender_email",
|
|
default=self._defaults["sender_email"],
|
|
comment="Sender email address",
|
|
)
|
|
section.add_option(
|
|
StringOption, "encoding", default=self._defaults["encoding"], comment="Email encoding"
|
|
)
|
|
section.add_option(
|
|
StringOption,
|
|
"catch_all_addr",
|
|
default=self._defaults["catch_all_addr"],
|
|
comment="Catch all sent emails to this specified email address",
|
|
)
|
|
|
|
section.add_option(
|
|
StringOption,
|
|
"templates_path",
|
|
default=self._defaults["templates_path"],
|
|
comment="Path to templates directory",
|
|
)
|
|
|
|
return section
|
|
|
|
def initialize(self, *args, **kwargs): # pylint: disable=arguments-differ
|
|
"""Configuration initialized hook"""
|
|
super().initialize(*args, **kwargs)
|
|
self.load_templates_directory()
|
|
|
|
def load_templates_directory(self, templates_path=None):
|
|
"""Load templates from specified directory"""
|
|
if templates_path is None:
|
|
templates_path = self._get_option("templates_path")
|
|
if not templates_path:
|
|
return
|
|
log.debug("Load email templates from %s directory", templates_path)
|
|
for filename in os.listdir(templates_path):
|
|
filepath = os.path.join(templates_path, filename)
|
|
if not os.path.isfile(filepath):
|
|
continue
|
|
template_name, template_type = os.path.splitext(filename)
|
|
if template_type not in [".html", ".txt", ".subject"]:
|
|
continue
|
|
template_type = "text" if template_type == ".txt" else template_type[1:]
|
|
if template_name not in self.templates:
|
|
self.templates[template_name] = {"path": templates_path}
|
|
log.debug("Load email template %s %s from %s", template_name, template_type, filepath)
|
|
with open(filepath, encoding="utf8") as file_desc:
|
|
self.templates[template_name][template_type] = MakoTemplate(
|
|
file_desc.read()
|
|
) # nosec
|
|
|
|
def forge_message(
|
|
self,
|
|
recipients,
|
|
subject=None,
|
|
html_body=None,
|
|
text_body=None, # pylint: disable=too-many-arguments,too-many-locals
|
|
attachment_files=None,
|
|
attachment_payloads=None,
|
|
sender_name=None,
|
|
sender_email=None,
|
|
encoding=None,
|
|
template=None,
|
|
cc=None,
|
|
**template_vars,
|
|
):
|
|
"""
|
|
Forge a message
|
|
|
|
:param recipients: The recipient(s) of the email. List of tuple(name, email) or
|
|
just the email of the recipients.
|
|
:param subject: The subject of the email.
|
|
:param html_body: The HTML body of the email
|
|
:param text_body: The plain text body of the email
|
|
:param attachment_files: List of filepaths to attach
|
|
:param attachment_payloads: List of tuples with filename and payload to attach
|
|
:param sender_name: Custom sender name (default: as defined on initialization)
|
|
:param sender_email: Custom sender email (default: as defined on initialization)
|
|
:param encoding: Email content encoding (default: as defined on initialization)
|
|
:param template: The name of a template to use to forge this email
|
|
:param cc: Optional list of CC recipient addresses.
|
|
List of tuple(name, email) or just the email of the recipients.
|
|
|
|
All other parameters will be consider as template variables.
|
|
"""
|
|
recipients = [recipients] if not isinstance(recipients, list) else recipients
|
|
msg = MIMEMultipart("alternative")
|
|
msg["To"] = ", ".join(
|
|
[
|
|
email.utils.formataddr(recipient) if isinstance(recipient, tuple) else recipient
|
|
for recipient in recipients
|
|
]
|
|
)
|
|
|
|
if cc:
|
|
cc = [cc] if not isinstance(cc, list) else cc
|
|
msg["Cc"] = ", ".join(
|
|
[
|
|
email.utils.formataddr(recipient) if isinstance(recipient, tuple) else recipient
|
|
for recipient in cc
|
|
]
|
|
)
|
|
|
|
msg["From"] = email.utils.formataddr(
|
|
(
|
|
sender_name or self._get_option("sender_name"),
|
|
sender_email or self._get_option("sender_email"),
|
|
)
|
|
)
|
|
if subject:
|
|
msg["Subject"] = (
|
|
subject.render(**template_vars)
|
|
if isinstance(subject, MakoTemplate)
|
|
else subject.format(**template_vars)
|
|
)
|
|
msg["Date"] = email.utils.formatdate(None, True)
|
|
encoding = encoding if encoding else self._get_option("encoding")
|
|
if template:
|
|
log.debug("Forge email from template %s", template)
|
|
assert template in self.templates, f"Unknown template {template}"
|
|
# Handle subject from template
|
|
if not subject:
|
|
assert self.templates[template].get(
|
|
"subject"
|
|
), f"No subject defined in template {template}"
|
|
msg["Subject"] = (
|
|
self.templates[template]["subject"].render(**template_vars)
|
|
if isinstance(self.templates[template]["subject"], MakoTemplate)
|
|
else self.templates[template]["subject"].format(**template_vars)
|
|
)
|
|
|
|
# Put HTML part in last one to preferred it
|
|
parts = []
|
|
if self.templates[template].get("text"):
|
|
if isinstance(self.templates[template]["text"], MakoTemplate):
|
|
parts.append(
|
|
(self.templates[template]["text"].render(**template_vars), "plain")
|
|
)
|
|
else:
|
|
parts.append(
|
|
(self.templates[template]["text"].format(**template_vars), "plain")
|
|
)
|
|
if self.templates[template].get("html"):
|
|
if isinstance(self.templates[template]["html"], MakoTemplate):
|
|
template_vars["load_image_as_base64"] = self.template_image_loader(
|
|
self.templates[template].get("path")
|
|
)
|
|
parts.append((self.templates[template]["html"].render(**template_vars), "html"))
|
|
else:
|
|
parts.append((self.templates[template]["html"].format(**template_vars), "html"))
|
|
|
|
for body, mime_type in parts:
|
|
msg.attach(MIMEText(body.encode(encoding), mime_type, _charset=encoding))
|
|
else:
|
|
assert subject, "No subject provided"
|
|
if text_body:
|
|
msg.attach(MIMEText(text_body.encode(encoding), "plain", _charset=encoding))
|
|
if html_body:
|
|
msg.attach(MIMEText(html_body.encode(encoding), "html", _charset=encoding))
|
|
if attachment_files:
|
|
for filepath in attachment_files:
|
|
with open(filepath, "rb") as fp:
|
|
part = MIMEBase("application", "octet-stream")
|
|
part.set_payload(fp.read())
|
|
encode_base64(part)
|
|
part.add_header(
|
|
"Content-Disposition",
|
|
f'attachment; filename="{os.path.basename(filepath)}"',
|
|
)
|
|
msg.attach(part)
|
|
if attachment_payloads:
|
|
for filename, payload in attachment_payloads:
|
|
part = MIMEBase("application", "octet-stream")
|
|
part.set_payload(payload)
|
|
encode_base64(part)
|
|
part.add_header("Content-Disposition", f'attachment; filename="{filename}"')
|
|
msg.attach(part)
|
|
return msg
|
|
|
|
@staticmethod
|
|
def template_image_loader(directory_path):
|
|
"""Return wrapper for the load_image_as_base64 function bind on the template directory"""
|
|
|
|
def _load_image_as_base64(path):
|
|
return load_image_as_base64(
|
|
os.path.join(directory_path, path)
|
|
if directory_path and not os.path.isabs(path)
|
|
else path
|
|
)
|
|
|
|
return _load_image_as_base64
|
|
|
|
def send(
|
|
self, recipients, msg=None, subject=None, just_try=None, cc=None, bcc=None, **forge_args
|
|
):
|
|
"""
|
|
Send an email
|
|
|
|
:param recipients: The recipient(s) of the email. List of tuple(name, email) or
|
|
just the email of the recipients.
|
|
:param msg: The message of this email (as MIMEBase or derivated classes)
|
|
:param subject: The subject of the email (only if the message is not provided
|
|
using msg parameter)
|
|
:param just_try: Enable just try mode (do not really send email, default: as defined on
|
|
initialization)
|
|
:param cc: Optional list of CC recipient addresses. List of tuple(name, email) or
|
|
just the email of the recipients.
|
|
:param bcc: Optional list of BCC recipient addresses. List of tuple(name, email) or
|
|
just the email of the recipients.
|
|
|
|
All other parameters will be consider as parameters to forge the message
|
|
(only if the message is not provided using msg parameter).
|
|
"""
|
|
recipients = [recipients] if not isinstance(recipients, list) else recipients
|
|
msg = msg if msg else self.forge_message(recipients, subject, cc=cc, **forge_args)
|
|
catch_addr = self._get_option("catch_all_addr")
|
|
if catch_addr:
|
|
log.debug(
|
|
"Catch email originally send to %s (CC:%s, BCC:%s) to %s",
|
|
", ".join(recipients),
|
|
", ".join(cc) if isinstance(cc, list) else cc,
|
|
", ".join(bcc) if isinstance(bcc, list) else bcc,
|
|
catch_addr,
|
|
)
|
|
recipients = catch_addr if isinstance(catch_addr, list) else [catch_addr]
|
|
else:
|
|
if cc:
|
|
recipients.extend(
|
|
[
|
|
recipient[1] if isinstance(recipient, tuple) else recipient
|
|
for recipient in (cc if isinstance(cc, list) else [cc])
|
|
]
|
|
)
|
|
if bcc:
|
|
recipients.extend(
|
|
[
|
|
recipient[1] if isinstance(recipient, tuple) else recipient
|
|
for recipient in (bcc if isinstance(bcc, list) else [bcc])
|
|
]
|
|
)
|
|
|
|
if just_try if just_try is not None else self._just_try:
|
|
log.debug(
|
|
'Just-try mode: do not really send this email to %s (subject="%s")',
|
|
", ".join(recipients),
|
|
subject or msg.get("subject", "No subject"),
|
|
)
|
|
return True
|
|
|
|
smtp_host = self._get_option("smtp_host")
|
|
smtp_port = self._get_option("smtp_port")
|
|
try:
|
|
if self._get_option("smtp_ssl"):
|
|
logging.info("Establish SSL connection to server %s:%s", smtp_host, smtp_port)
|
|
server = smtplib.SMTP_SSL(smtp_host, smtp_port)
|
|
else:
|
|
logging.info("Establish connection to server %s:%s", smtp_host, smtp_port)
|
|
server = smtplib.SMTP(smtp_host, smtp_port)
|
|
if self._get_option("smtp_tls"):
|
|
logging.info("Start TLS on SMTP connection")
|
|
server.starttls()
|
|
except smtplib.SMTPException:
|
|
log.error("Error connecting to SMTP server %s:%s", smtp_host, smtp_port, exc_info=True)
|
|
return False
|
|
|
|
if self._get_option("smtp_debug"):
|
|
server.set_debuglevel(True)
|
|
|
|
smtp_user = self._get_option("smtp_user")
|
|
smtp_password = self._get_option("smtp_password")
|
|
if smtp_user and smtp_password:
|
|
try:
|
|
log.info("Try to authenticate on SMTP connection as %s", smtp_user)
|
|
server.login(smtp_user, smtp_password)
|
|
except smtplib.SMTPException:
|
|
log.error(
|
|
"Error authenticating on SMTP server %s:%s with user %s",
|
|
smtp_host,
|
|
smtp_port,
|
|
smtp_user,
|
|
exc_info=True,
|
|
)
|
|
return False
|
|
|
|
error = False
|
|
try:
|
|
log.info("Sending email to %s", ", ".join(recipients))
|
|
server.sendmail(
|
|
self._get_option("sender_email"),
|
|
[
|
|
recipient[1] if isinstance(recipient, tuple) else recipient
|
|
for recipient in recipients
|
|
],
|
|
msg.as_string(),
|
|
)
|
|
except smtplib.SMTPException:
|
|
error = True
|
|
log.error("Error sending email to %s", ", ".join(recipients), exc_info=True)
|
|
finally:
|
|
server.quit()
|
|
|
|
return not error
|