python-mylib/mylib/email.py

595 lines
21 KiB
Python
Raw Normal View History

2021-05-19 19:19:57 +02:00
""" Email client to forge and send emails """
2021-03-24 19:20:36 +01:00
import email.utils
2021-01-13 16:09:33 +01:00
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
2021-01-13 16:09:33 +01:00
from mako.template import Template as MakoTemplate
from mylib.config import (
BooleanOption,
ConfigurableObject,
IntegerOption,
PasswordOption,
StringOption,
)
2021-01-13 16:09:33 +01:00
log = logging.getLogger(__name__)
class EmailClient(
ConfigurableObject
): # pylint: disable=useless-object-inheritance,too-many-instance-attributes
2021-01-13 16:09:33 +01:00
"""
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,
}
2021-01-13 16:09:33 +01:00
2023-01-06 19:36:14 +01:00
templates = {}
2021-01-13 16:09:33 +01:00
def __init__(self, templates=None, initialize=False, **kwargs):
super().__init__(**kwargs)
2021-01-13 16:09:33 +01:00
assert templates is None or isinstance(templates, dict)
2023-01-06 19:36:14 +01:00
self.templates = templates if templates else {}
if initialize:
self.initialize()
2021-01-13 16:09:33 +01:00
2023-01-06 19:36:14 +01:00
# 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] = {}
log.debug("Load email template %s %s from %s", template_name, template_type, filepath)
with open(filepath, encoding="utf8") as file_desc:
2023-03-23 09:56:31 +01:00
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,
):
2021-01-13 16:09:33 +01:00
"""
Forge a message
:param recipients: The recipient(s) of the email. List of tuple(name, email) or
just the email of the recipients.
2021-01-13 16:09:33 +01:00
: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
2021-01-13 16:09:33 +01:00
: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.
2021-01-13 16:09:33 +01:00
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"),
)
)
2021-01-13 16:09:33 +01:00
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")
2021-01-13 16:09:33 +01:00
if template:
assert template in self.templates, f"Unknwon template {template}"
2021-01-13 16:09:33 +01:00
# 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)
)
2021-01-13 16:09:33 +01:00
# Put HTML part in last one to prefered 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):
parts.append((self.templates[template]["html"].render(**template_vars), "html"))
else:
parts.append((self.templates[template]["html"].format(**template_vars), "html"))
2021-01-13 16:09:33 +01:00
for body, mime_type in parts:
msg.attach(MIMEText(body.encode(encoding), mime_type, _charset=encoding))
2021-01-13 16:09:33 +01:00
else:
assert subject, "No subject provided"
2021-01-13 16:09:33 +01:00
if text_body:
msg.attach(MIMEText(text_body.encode(encoding), "plain", _charset=encoding))
2021-01-13 16:09:33 +01:00
if html_body:
msg.attach(MIMEText(html_body.encode(encoding), "html", _charset=encoding))
2021-01-13 16:09:33 +01:00
if attachment_files:
for filepath in attachment_files:
with open(filepath, "rb") as fp:
part = MIMEBase("application", "octet-stream")
2021-01-13 16:09:33 +01:00
part.set_payload(fp.read())
encode_base64(part)
2023-01-06 22:13:28 +01:00
part.add_header(
"Content-Disposition",
f'attachment; filename="{os.path.basename(filepath)}"',
)
2021-01-13 16:09:33 +01:00
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)
2021-01-13 16:09:33 +01:00
return msg
def send(
self, recipients, msg=None, subject=None, just_try=None, cc=None, bcc=None, **forge_args
):
2021-01-13 16:09:33 +01:00
"""
Send an email
:param recipients: The recipient(s) of the email. List of tuple(name, email) or
just the email of the recipients.
2021-01-13 16:09:33 +01:00
: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.
2021-01-13 16:09:33 +01:00
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 originaly 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])
]
)
2021-01-13 16:09:33 +01:00
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"),
)
2021-01-13 16:09:33 +01:00
return True
smtp_host = self._get_option("smtp_host")
smtp_port = self._get_option("smtp_port")
2021-01-13 16:09:33 +01:00
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)
2021-01-13 16:09:33 +01:00
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")
2021-01-13 16:09:33 +01:00
server.starttls()
except smtplib.SMTPException:
log.error("Error connecting to SMTP server %s:%s", smtp_host, smtp_port, exc_info=True)
2021-01-13 16:09:33 +01:00
return False
if self._get_option("smtp_debug"):
2021-01-13 16:09:33 +01:00
server.set_debuglevel(True)
smtp_user = self._get_option("smtp_user")
smtp_password = self._get_option("smtp_password")
if smtp_user and smtp_password:
2021-01-13 16:09:33 +01:00
try:
log.info("Try to authenticate on SMTP connection as %s", smtp_user)
server.login(smtp_user, smtp_password)
2021-01-13 16:09:33 +01:00
except smtplib.SMTPException:
log.error(
"Error authenticating on SMTP server %s:%s with user %s",
smtp_host,
smtp_port,
smtp_user,
exc_info=True,
)
2021-01-13 16:09:33 +01:00
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(),
)
2021-01-13 16:09:33 +01:00
except smtplib.SMTPException:
error = True
log.error("Error sending email to %s", ", ".join(recipients), exc_info=True)
2021-01-13 16:09:33 +01:00
finally:
server.quit()
return not error
if __name__ == "__main__":
2021-03-24 19:20:36 +01:00
# Run tests
import argparse
2021-01-13 16:09:33 +01:00
import datetime
import sys
# Options parser
parser = argparse.ArgumentParser()
parser.add_argument(
"-v", "--verbose", action="store_true", dest="verbose", help="Enable verbose mode"
2021-01-13 16:09:33 +01:00
)
parser.add_argument(
"-d", "--debug", action="store_true", dest="debug", help="Enable debug mode"
2021-01-13 16:09:33 +01:00
)
parser.add_argument(
"-l", "--log-file", action="store", type=str, dest="logfile", help="Log file path"
2021-01-13 16:09:33 +01:00
)
parser.add_argument(
"-j", "--just-try", action="store_true", dest="just_try", help="Enable just-try mode"
2021-01-13 16:09:33 +01:00
)
email_opts = parser.add_argument_group("Email options")
2021-01-13 16:09:33 +01:00
email_opts.add_argument(
"-H", "--smtp-host", action="store", type=str, dest="email_smtp_host", help="SMTP host"
2021-01-13 16:09:33 +01:00
)
email_opts.add_argument(
"-P", "--smtp-port", action="store", type=int, dest="email_smtp_port", help="SMTP port"
2021-01-13 16:09:33 +01:00
)
email_opts.add_argument(
"-S", "--smtp-ssl", action="store_true", dest="email_smtp_ssl", help="Use SSL"
2021-01-13 16:09:33 +01:00
)
email_opts.add_argument(
"-T", "--smtp-tls", action="store_true", dest="email_smtp_tls", help="Use TLS"
2021-01-13 16:09:33 +01:00
)
email_opts.add_argument(
"-u", "--smtp-user", action="store", type=str, dest="email_smtp_user", help="SMTP username"
2021-01-13 16:09:33 +01:00
)
email_opts.add_argument(
"-p",
"--smtp-password",
2021-01-13 16:09:33 +01:00
action="store",
type=str,
dest="email_smtp_password",
help="SMTP password",
2021-01-13 16:09:33 +01:00
)
email_opts.add_argument(
"-D",
"--smtp-debug",
2021-01-13 16:09:33 +01:00
action="store_true",
dest="email_smtp_debug",
help="Debug SMTP connection",
2021-01-13 16:09:33 +01:00
)
email_opts.add_argument(
"-e",
"--email-encoding",
2021-01-13 16:09:33 +01:00
action="store",
type=str,
dest="email_encoding",
help="SMTP encoding",
2021-01-13 16:09:33 +01:00
)
email_opts.add_argument(
"-f",
"--sender-name",
2021-01-13 16:09:33 +01:00
action="store",
type=str,
dest="email_sender_name",
help="Sender name",
2021-01-13 16:09:33 +01:00
)
email_opts.add_argument(
"-F",
"--sender-email",
2021-01-13 16:09:33 +01:00
action="store",
type=str,
dest="email_sender_email",
help="Sender email",
2021-01-13 16:09:33 +01:00
)
email_opts.add_argument(
"-C",
"--catch-all",
2021-01-13 16:09:33 +01:00
action="store",
type=str,
dest="email_catch_all",
help="Catch all sent email: specify catch recipient email address",
2021-01-13 16:09:33 +01:00
)
test_opts = parser.add_argument_group("Test email options")
2021-01-13 16:09:33 +01:00
test_opts.add_argument(
"-t",
"--to",
2021-01-13 16:09:33 +01:00
action="store",
type=str,
dest="test_to",
help="Test email recipient",
)
test_opts.add_argument(
"-m",
"--mako",
action="store_true",
dest="test_mako",
help="Test mako templating",
)
2021-01-13 16:09:33 +01:00
options = parser.parse_args()
if not options.test_to:
parser.error("You must specify test email recipient using -t/--to parameter")
2021-01-13 16:09:33 +01:00
sys.exit(1)
# Initialize logs
logformat = "%(asctime)s - Test EmailClient - %(levelname)s - %(message)s"
2021-01-13 16:09:33 +01:00
if options.debug:
loglevel = logging.DEBUG
elif options.verbose:
loglevel = logging.INFO
else:
loglevel = logging.WARNING
if options.logfile:
logging.basicConfig(filename=options.logfile, level=loglevel, format=logformat)
else:
logging.basicConfig(level=loglevel, format=logformat)
if options.email_smtp_user and not options.email_smtp_password:
import getpass
options.email_smtp_password = getpass.getpass("Please enter SMTP password: ")
logging.info("Initialize Email client")
2021-01-13 16:09:33 +01:00
email_client = EmailClient(
smtp_host=options.email_smtp_host,
smtp_port=options.email_smtp_port,
smtp_ssl=options.email_smtp_ssl,
smtp_tls=options.email_smtp_tls,
smtp_user=options.email_smtp_user,
smtp_password=options.email_smtp_password,
smtp_debug=options.email_smtp_debug,
sender_name=options.email_sender_name,
sender_email=options.email_sender_email,
catch_all_addr=options.email_catch_all,
just_try=options.just_try,
encoding=options.email_encoding,
2021-03-24 19:20:36 +01:00
templates=dict(
test=dict(
subject="Test email",
text=(
"Just a test email sent at {sent_date}."
if not options.test_mako
2023-03-23 09:56:31 +01:00
else MakoTemplate("Just a test email sent at ${sent_date | h}.") # nosec
2021-03-24 19:20:36 +01:00
),
html=(
2023-03-23 09:56:31 +01:00
"<strong>Just a test email.</strong> <small>(sent at {sent_date | h})</small>"
if not options.test_mako
2023-03-23 09:56:31 +01:00
else MakoTemplate( # nosec
"<strong>Just a test email.</strong> "
"<small>(sent at ${sent_date | h})</small>"
)
),
2021-03-24 19:20:36 +01:00
)
),
2021-01-13 16:09:33 +01:00
)
logging.info("Send a test email to %s", options.test_to)
if email_client.send(options.test_to, template="test", sent_date=datetime.datetime.now()):
logging.info("Test email sent")
2021-01-13 16:09:33 +01:00
sys.exit(0)
logging.error("Fail to send test email")
2021-01-13 16:09:33 +01:00
sys.exit(1)