python-mylib/mylib/email.py

456 lines
16 KiB
Python
Raw Normal View History

2021-01-13 16:09:33 +01:00
# -*- coding: utf-8 -*-
2021-05-19 19:19:57 +02:00
""" Email client to forge and send emails """
2021-03-24 19:20:36 +01:00
2021-01-13 16:09:33 +01:00
import logging
import os
import smtplib
import email.utils
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.mime.base import MIMEBase
from email.encoders import encode_base64
from mako.template import Template as MakoTemplate
from mylib.config import ConfigurableObject
from mylib.config import BooleanOption
from mylib.config import IntegerOption
from mylib.config import PasswordOption
from mylib.config import 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,
}
2021-01-13 16:09:33 +01:00
templates = dict()
def __init__(self, templates=None, **kwargs):
super().__init__(**kwargs)
2021-01-13 16:09:33 +01:00
assert templates is None or isinstance(templates, dict)
self.templates = templates if templates else dict()
def configure(self, use_smtp=True, just_try=True, ** kwargs): # pylint: disable=arguments-differ
""" Configure options on registered mylib.Config object """
section = super().configure(**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')
if just_try:
section.add_option(
BooleanOption, 'just_try', default=self._defaults['just_try'],
comment='Just-try mode: do not really send emails')
return section
def forge_message(self, rcpt_to, 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, **template_vars):
2021-01-13 16:09:33 +01:00
"""
Forge a message
:param rcpt_to: The recipient of the email. Could be a tuple(name, email) or just the email of the recipient.
: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
All other parameters will be consider as template variables.
"""
msg = MIMEMultipart('alternative')
msg['To'] = email.utils.formataddr(rcpt_to) if isinstance(rcpt_to, tuple) else rcpt_to
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.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, "Unknwon template %s" % template
2021-01-13 16:09:33 +01:00
# Handle subject from template
if not subject:
assert self.templates[template].get('subject'), 'No subject defined in template %s' % template
msg['Subject'] = 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'
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', 'attachment; filename="%s"' % 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', 'attachment; filename="%s"' % filename)
msg.attach(part)
2021-01-13 16:09:33 +01:00
return msg
def send(self, rcpt_to, msg=None, subject=None, just_try=False, **forge_args):
"""
Send an email
:param rcpt_to: The recipient of the email. Could be a tuple(name, email)
or just the email of the recipient.
: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)
All other parameters will be consider as parameters to forge the message
(only if the message is not provided using msg parameter).
"""
msg = msg if msg else self.forge_message(rcpt_to, subject, **forge_args)
if just_try or self._get_option('just_try'):
2021-01-13 16:09:33 +01:00
log.debug('Just-try mode: do not really send this email to %s (subject="%s")', rcpt_to, subject or msg.get('subject', 'No subject'))
return True
catch_addr = self._get_option('catch_all_addr')
if catch_addr:
2021-01-13 16:09:33 +01:00
log.debug('Catch email originaly send to %s to %s', rcpt_to, catch_addr)
rcpt_to = catch_addr
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'):
2021-01-13 16:09:33 +01:00
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)
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', rcpt_to)
server.sendmail(
self._get_option('sender_email'),
[rcpt_to[1] if isinstance(rcpt_to, tuple) else rcpt_to],
msg.as_string()
)
2021-01-13 16:09:33 +01:00
except smtplib.SMTPException:
error = True
log.error('Error sending email to %s', rcpt_to, exc_info=True)
finally:
server.quit()
return not error
if __name__ == '__main__':
2021-03-24 19:20:36 +01:00
# Run tests
2021-01-13 16:09:33 +01:00
import datetime
import sys
import argparse
# Options parser
parser = argparse.ArgumentParser()
parser.add_argument(
'-v', '--verbose',
action="store_true",
dest="verbose",
help="Enable verbose mode"
)
parser.add_argument(
'-d', '--debug',
action="store_true",
dest="debug",
help="Enable debug mode"
)
parser.add_argument(
'-l', '--log-file',
action="store",
type=str,
dest="logfile",
help="Log file path"
)
parser.add_argument(
'-j', '--just-try',
action="store_true",
dest="just_try",
help="Enable just-try mode"
)
email_opts = parser.add_argument_group('Email options')
email_opts.add_argument(
'-H', '--smtp-host',
action="store",
type=str,
dest="email_smtp_host",
help="SMTP host"
)
email_opts.add_argument(
'-P', '--smtp-port',
action="store",
type=int,
dest="email_smtp_port",
help="SMTP port"
)
email_opts.add_argument(
'-S', '--smtp-ssl',
action="store_true",
dest="email_smtp_ssl",
help="Use SSL"
)
email_opts.add_argument(
'-T', '--smtp-tls',
action="store_true",
dest="email_smtp_tls",
help="Use TLS"
)
email_opts.add_argument(
'-u', '--smtp-user',
action="store",
type=str,
dest="email_smtp_user",
help="SMTP username"
)
email_opts.add_argument(
'-p', '--smtp-password',
action="store",
type=str,
dest="email_smtp_password",
help="SMTP password"
)
email_opts.add_argument(
'-D', '--smtp-debug',
action="store_true",
dest="email_smtp_debug",
help="Debug SMTP connection"
)
email_opts.add_argument(
'-e', '--email-encoding',
action="store",
type=str,
dest="email_encoding",
help="SMTP encoding"
)
email_opts.add_argument(
'-f', '--sender-name',
action="store",
type=str,
dest="email_sender_name",
help="Sender name"
)
email_opts.add_argument(
'-F', '--sender-email',
action="store",
type=str,
dest="email_sender_email",
help="Sender email"
)
email_opts.add_argument(
'-C', '--catch-all',
action="store",
type=str,
dest="email_catch_all",
help="Catch all sent email: specify catch recipient email address"
)
test_opts = parser.add_argument_group('Test email options')
test_opts.add_argument(
'-t', '--to',
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')
sys.exit(1)
# Initialize logs
logformat = '%(asctime)s - Test EmailClient - %(levelname)s - %(message)s'
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')
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 else
MakoTemplate("Just a test email sent at ${sent_date}.")
),
html=(
2021-03-24 19:36:07 +01:00
"<strong>Just a test email.</strong> <small>(sent at {sent_date})</small>" if not options.test_mako else
MakoTemplate("<strong>Just a test email.</strong> <small>(sent at ${sent_date})</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')
sys.exit(0)
logging.error('Fail to send test email')
sys.exit(1)