102 lines
3.3 KiB
Python
102 lines
3.3 KiB
Python
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
import email
|
|
import enum
|
|
import logging
|
|
import smtplib
|
|
|
|
from multischleuder.reporting import Message
|
|
|
|
|
|
class TlsMode(enum.Enum):
|
|
PLAIN = 'smtp', 25
|
|
SMTPS = 'smtps', 465
|
|
STARTTLS = 'smtp+starttls', 587
|
|
|
|
def __init__(self, proto: str, port: int):
|
|
self._proto = proto
|
|
self._port = port
|
|
|
|
@property
|
|
def proto(self):
|
|
return self._proto
|
|
|
|
@property
|
|
def port(self):
|
|
return self._port
|
|
|
|
|
|
class SmtpClient:
|
|
|
|
def __init__(self,
|
|
hostname: str = 'localhost',
|
|
port: int = 25,
|
|
tls: 'TlsMode' = TlsMode.PLAIN,
|
|
username: Optional[str] = None,
|
|
password: Optional[str] = None):
|
|
self._hostname: str = hostname
|
|
self._port: int = port
|
|
self._tls: TlsMode = tls
|
|
self._username: Optional[str] = username
|
|
self._password: Optional[str] = password
|
|
self._smtp: Optional[smtplib.SMTP] = None
|
|
self._dry_run: bool = False
|
|
self._logger = logging.getLogger()
|
|
|
|
@staticmethod
|
|
def parse(config: Dict[str, Any]) -> 'SmtpClient':
|
|
tls = TlsMode[config.get('tls', 'PLAIN')]
|
|
return SmtpClient(
|
|
hostname=config.get('hostname', 'localhost'),
|
|
port=config.get('port', tls.port),
|
|
tls=tls,
|
|
username=config.get('username'),
|
|
password=config.get('password')
|
|
)
|
|
|
|
def dry_run(self):
|
|
self._dry_run = True
|
|
|
|
def send_messages(self, messages: List[Message]):
|
|
if len(messages) > 0 and not self._dry_run:
|
|
with self as smtp:
|
|
for m in messages:
|
|
msg = m.mime
|
|
self._logger.debug(f'MIME Message:\n{str(msg)}')
|
|
self._send_message(msg)
|
|
|
|
def _send_message(self, msg: email.message.Message):
|
|
if self._smtp is None:
|
|
raise RuntimeError('SMTP connection is not established')
|
|
if not self._dry_run:
|
|
self._smtp.send_message(msg)
|
|
self._logger.debug(f'Sent email message.')
|
|
|
|
def __enter__(self):
|
|
# TLS from the start requires a different class
|
|
cls = smtplib.SMTP if self._tls != TlsMode.SMTPS else smtplib.SMTP_SSL
|
|
self._smtp = cls(self._hostname, self._port)
|
|
# Establish the connection
|
|
smtp = self._smtp.__enter__()
|
|
if self._tls == TlsMode.STARTTLS:
|
|
smtp.starttls()
|
|
# Only sign in if both username and password are provided
|
|
if self._username is not None and self._password is not None:
|
|
smtp.login(self._username, self._password)
|
|
self._logger.debug(f'SMTP connection to {str(self)} established')
|
|
return smtp
|
|
|
|
def __exit__(self, exc_type, exc_val, exc_tb):
|
|
if self._smtp is None:
|
|
raise RuntimeError('SMTP connection is not established')
|
|
self._smtp.quit()
|
|
ret = self._smtp.__exit__(exc_type, exc_val, exc_tb)
|
|
self._logger.debug(f'SMTP connection to {str(self)} closed')
|
|
self._smtp = None
|
|
return ret
|
|
|
|
def __str__(self) -> str:
|
|
if self._username is not None and self._password is not None:
|
|
return f'{self._tls.proto}://{self._username}@{self._hostname}:{self._port}'
|
|
return f'{self._tls.proto}://{self._hostname}:{self._port}'
|