import logging
import socket
import smtplib
from ssl import SSLError, CertificateError, create_default_context, CERT_NONE
from typing import List, Dict, Tuple
from mailsuite.utils import create_email
logger = logging.getLogger(__name__)
[docs]class SMTPError(RuntimeError):
"""Raised when a SMTP error occurs"""
[docs]def send_email(host: str, message_from: str, message_to: List[str] = None,
message_cc: List = None, message_bcc: List = None,
port: int = 0, require_encryption: bool = False,
verify: bool = True, username: str = None, password: str = None,
envelope_from: str = None, subject: str = None,
message_headers: Dict = None,
attachments: Tuple[str, bytes] = None,
plain_message: str = None, html_message: str = None):
"""
Send an email using a SMTP relay
Args:
host: Mail server hostname or IP address
message_from: The value of the message from header
message_to: A list of addresses to send mail to
message_cc: A List of addresses to Carbon Copy (CC)
message_bcc: A list of addresses to Blind Carbon Copy (BCC)
port: Port to use
require_encryption: Require a SSL/TLS connection from the start
verify: Verify the SSL/TLS certificate
username: An optional username
password: An optional password
envelope_from: Overrides the SMTP envelope "mail from" header
subject: The message subject
message_headers: Custom message headers
attachments: A list of tuples, containing filenames and bytes
plain_message: The plain text message body
html_message: The HTML message body
"""
msg = create_email(message_from=message_from, message_to=message_to,
message_cc=message_cc, subject=subject,
message_headers=message_headers,
attachments=attachments,
plain_message=plain_message, html_message=html_message)
try:
ssl_context = create_default_context()
if verify is False:
ssl_context.check_hostname = False
ssl_context.verify_mode = CERT_NONE
if require_encryption:
server = smtplib.SMTP_SSL(host, port=port, context=ssl_context)
server.connect(host, port)
server.ehlo_or_helo_if_needed()
else:
server = smtplib.SMTP(host, port=port)
server.connect(host, port)
server.ehlo_or_helo_if_needed()
if server.has_extn("starttls"):
server.starttls(context=ssl_context)
server.ehlo()
else:
logger.warning("SMTP server does not support STARTTLS. "
"Proceeding in plain text!")
if username and password:
server.login(username, password)
if envelope_from is None:
envelope_from = message_from
envelope_to = message_to.copy()
if message_cc is not None:
message_to += message_cc
if message_bcc is not None:
message_to += message_bcc
envelope_to = list(set(envelope_to))
server.sendmail(envelope_from, envelope_to, msg)
except smtplib.SMTPException as error:
error = error.__str__().lstrip("b'").rstrip("'").rstrip(".")
raise SMTPError(error)
except socket.gaierror:
raise SMTPError("DNS resolution failed")
except ConnectionRefusedError:
raise SMTPError("Connection refused")
except ConnectionResetError:
raise SMTPError("Connection reset")
except ConnectionAbortedError:
raise SMTPError("Connection aborted")
except TimeoutError:
raise SMTPError("Connection timed out")
except SSLError as error:
raise SMTPError("SSL error: {0}".format(error.__str__()))
except CertificateError as error:
raise SMTPError("Certificate error: {0}".format(error.__str__()))