lukmangrd/platypush
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

Added SMTP plugin [links to #146]

Browse source
This commit is contained in:
Fabio Manganiello 2020-09-01 01:52:22 +02:00
parent 737c135996
commit 681e9f1703
2 changed files with 189 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

99
platypush/plugins/mail/__init__.py
View file

@ -5,6 +5,14 @@ import subprocess
from abc import ABC
from dataclasses import dataclass
from datetime import datetime
from email.message import Message
from email.mime.application import MIMEApplication
from email.mime.audio import MIMEAudio
from email.mime.base import MIMEBase
from email.mime.image import MIMEImage
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from mimetypes import guess_type
from typing import Optional, List, Union, Any, Dict
from platypush.message import JSONAble
@ -130,4 +138,95 @@ class MailInPlugin(MailPlugin, ABC):
raise NotImplementedError()
class MailOutPlugin(MailPlugin, ABC):
"""
Base class for mail out plugins.
"""
def send_message(self, message: Message, **connect_args):
raise NotImplementedError()
@staticmethod
def _file_to_part(file: str) -> MIMEBase:
_type, _subtype, _type_class = 'application', 'octet-stream', MIMEApplication
mime_type, _sub_subtype = guess_type(file)
if mime_type:
_type, _subtype = mime_type.split('/')
if _sub_subtype:
_subtype += ';' + _sub_subtype
if _type == 'application':
_type_class = MIMEApplication
elif _type == 'audio':
_type_class = MIMEAudio
elif _type == 'image':
_type_class = MIMEImage
elif _type == 'text':
_type_class = MIMEText
with open(file, 'rb') as f:
return _type_class(f.read(), _subtype, Name=os.path.basename(file))
@classmethod
def create_message(cls, to: Union[str, List[str]], from_: Optional[str] = None,
cc: Optional[Union[str, List[str]]] = None, bcc: Optional[Union[str, List[str]]] = None,
subject: str = '', body: str = '', body_type: str = 'plain',
attachments: Optional[List[str]] = None, headers: Optional[Dict[str, str]] = None) -> Message:
assert from_, 'from/from_ field not specified'
body = MIMEText(body, body_type)
if attachments:
msg = MIMEMultipart()
msg.attach(body)
for attachment in attachments:
attachment = os.path.abspath(os.path.expanduser(attachment))
assert os.path.isfile(attachment), 'No such file: {}'.format(attachment)
part = cls._file_to_part(attachment)
part['Content-Disposition'] = 'attachment; filename="{}"'.format(os.path.basename(attachment))
msg.attach(part)
else:
msg = body
msg['From'] = from_
msg['To'] = ', '.join(to) if isinstance(to, List) else to
msg['Cc'] = ', '.join(cc) if cc else ''
msg['Bcc'] = ', '.join(bcc) if bcc else ''
msg['Subject'] = subject
if headers:
for name, value in headers.items():
msg.add_header(name, value)
return msg
@action
def send(self, to: Union[str, List[str]], from_: Optional[str] = None,
cc: Optional[Union[str, List[str]]] = None, bcc: Optional[Union[str, List[str]]] = None,
subject: str = '', body: str = '', body_type: str = 'plain', attachments: Optional[List[str]] = None,
headers: Optional[Dict[str, str]] = None, **connect_args):
"""
Send an email through the specified SMTP sender.
:param to: Receiver(s), as comma-separated strings or list.
:param from_: Sender email address (``from`` is also supported outside of Python contexts).
:param cc: Carbon-copy addresses, as comma-separated strings or list
:param bcc: Blind carbon-copy addresses, as comma-separated strings or list
:param subject: Mail subject.
:param body: Mail body.
:param body_type: Mail body type, as a subtype of ``text/`` (e.g. ``html``) (default: ``plain``).
:param attachments: List of attachment files to send.
:param headers: Key-value map of headers to be added.
:param connect_args: Parameters for ``.connect()``, if you want to override the default server configuration.
"""
if not from_ and 'from' in connect_args:
from_ = connect_args.pop('from')
msg = self.create_message(to=to, from_=from_, cc=cc, bcc=bcc, subject=subject, body=body, body_type=body_type,
attachments=attachments, headers=headers)
return self.send_message(msg, **connect_args)
# vim:sw=4:ts=4:et:

90
platypush/plugins/mail/smtp.py Normal file
View file

@ -0,0 +1,90 @@
from email.message import Message
from typing import Optional, List
from smtplib import SMTP, SMTP_SSL
from platypush.plugins.mail import MailOutPlugin, ServerInfo
class MailSmtpPlugin(MailOutPlugin):
"""
Plugin to interact with a mail server over SMTP.
"""
_default_port = 25
_default_ssl_port = 465
def __init__(self, server: str, port: Optional[int] = None, local_hostname: Optional[str] = None,
source_address: Optional[List[str]] = None, username: Optional[str] = None,
password: Optional[str] = None, password_cmd: Optional[str] = None, access_token: Optional[str] = None,
oauth_mechanism: Optional[str] = 'XOAUTH2', oauth_vendor: Optional[str] = None, ssl: bool = False,
keyfile: Optional[str] = None, certfile: Optional[str] = None, timeout: Optional[int] = 60, **kwargs):
"""
:param server: Server name/address.
:param port: Port (default: 25 for plain, 465 for SSL).
:param local_hostname: If specified, local_hostname is used as the FQDN of the local host in the HELO/EHLO
command. Otherwise, the local hostname is found using socket.getfqdn().
:param source_address: The optional source_address parameter allows binding to some specific source address in
a machine with multiple network interfaces, and/or to some specific source TCP port. It takes a 2-tuple
(host, port), for the socket to bind to as its source address before connecting. If omitted (or if host or
port are '' and/or 0 respectively) the OS default behavior will be used.
:param username: SMTP username.
:param password: SMTP password.
:param password_cmd: If you don't want to input your password in the configuration, run this command to fetch
or decrypt the password.
:param access_token: OAuth2 access token if the server supports OAuth authentication.
:param oauth_mechanism: OAuth2 mechanism (default: ``XOAUTH2``).
:param oauth_vendor: OAuth2 vendor (default: None).
:param ssl: Use SSL (default: False).
:param keyfile: Private key file for SSL connection if client authentication is required.
:param certfile: SSL certificate file or chain.
:param timeout: Server connect/read timeout in seconds (default: 60).
"""
super().__init__(**kwargs)
self.local_hostname = local_hostname
self.source_address = source_address
self.server_info = self._get_server_info(server=server, port=port, username=username, password=password,
password_cmd=password_cmd, ssl=ssl, keyfile=keyfile, certfile=certfile,
access_token=access_token, oauth_mechanism=oauth_mechanism,
oauth_vendor=oauth_vendor, timeout=timeout)
def _get_server_info(self, server: Optional[str] = None, port: Optional[int] = None, username: Optional[str] = None,
password: Optional[str] = None, password_cmd: Optional[str] = None,
ssl: Optional[bool] = None, keyfile: Optional[str] = None, certfile: Optional[str] = None,
timeout: Optional[int] = None, **kwargs) -> ServerInfo:
return super()._get_server_info(server=server, port=port, username=username, password=password,
password_cmd=password_cmd, ssl=ssl, keyfile=keyfile, certfile=certfile,
default_port=self._default_port, default_ssl_port=self._default_ssl_port,
timeout=timeout)
def connect(self, **connect_args) -> SMTP:
info = self._get_server_info(**connect_args)
self.logger.info('Connecting to {}'.format(info.server))
smtp_args = {
'host': info.server,
'port': info.port,
'local_hostname': self.local_hostname,
'source_address': self.source_address,
}
if info.ssl:
client_type = SMTP_SSL
smtp_args.update(certfile=info.certfile, keyfile=info.keyfile)
else:
client_type = SMTP
client = client_type(**smtp_args)
if info.password:
client.login(info.username, info.password)
return client
def send_message(self, message: Message, **connect_args):
with self.connect(**connect_args) as client:
errors = client.sendmail(message['From'], message['To'], message.as_string())
if errors:
return None, ['{}: {}'.format(code, err) for code, err in errors.items()]
# vim:sw=4:ts=4:et: