86 lines
2.5 KiB
Python
86 lines
2.5 KiB
Python
import asyncio
|
|
import logging
|
|
import ssl
|
|
|
|
import aiosmtplib
|
|
from email.mime.base import MIMEBase
|
|
from email.mime.text import MIMEText
|
|
from email.mime.multipart import MIMEMultipart
|
|
from fastapi.responses import RedirectResponse
|
|
from typing import Union, List
|
|
|
|
|
|
async def send_email(
|
|
host: str,
|
|
port: int,
|
|
message: str,
|
|
timeout: int,
|
|
sender: str,
|
|
recipient: str,
|
|
subject: str,
|
|
login_email: str,
|
|
password: str,
|
|
attachments: List[MIMEBase] = tuple(),
|
|
use_tls: bool = True,
|
|
check_hostname: bool = True,
|
|
) -> Union[bool, aiosmtplib.SMTPException]:
|
|
"""
|
|
Sends an email with the given details. Returns True on success
|
|
or an exception object upon failure
|
|
"""
|
|
|
|
try:
|
|
async with aiosmtplib.SMTP(host, port, timeout=timeout) as srv:
|
|
msg = MIMEMultipart()
|
|
msg["From"] = sender
|
|
msg["To"] = recipient
|
|
msg["Subject"] = subject
|
|
msg.attach(MIMEText(message, "html"))
|
|
for attachment in attachments:
|
|
msg.attach(attachment)
|
|
await srv.ehlo() # We identify ourselves
|
|
if use_tls:
|
|
context = ssl.create_default_context()
|
|
context.check_hostname = check_hostname
|
|
await srv.starttls(tls_context=context)
|
|
await srv.ehlo() # We do it again, but encrypted!
|
|
await srv.login(login_email, password)
|
|
await srv.sendmail(sender, recipient, msg.as_string())
|
|
except (aiosmtplib.SMTPException, asyncio.TimeoutError) as error:
|
|
logging.error(f"An error occurred while dealing with {host}:{port} (SMTP): {type(error).__name__}: {error}")
|
|
return error
|
|
return True
|
|
|
|
|
|
async def test_smtp(
|
|
host: str,
|
|
port: int,
|
|
login_email: str,
|
|
password: str,
|
|
timeout: int,
|
|
use_tls: bool = True,
|
|
check_hostname: bool = True,
|
|
):
|
|
"""
|
|
Attempts login to the given SMTP server with the given credentials.
|
|
Used upon startup, raises an exception upon failure. This will
|
|
fail if the server does not support TLS encryption for login
|
|
"""
|
|
|
|
async with aiosmtplib.SMTP(host, port, timeout=timeout) as srv:
|
|
await srv.ehlo()
|
|
if use_tls:
|
|
context = ssl.create_default_context()
|
|
context.check_hostname = check_hostname
|
|
await srv.starttls(tls_context=context)
|
|
await srv.ehlo()
|
|
await srv.login(login_email, password)
|
|
|
|
|
|
def redirect(url: str) -> RedirectResponse:
|
|
"""
|
|
Returns a redirect response to the given url
|
|
"""
|
|
|
|
return RedirectResponse(url=url)
|