|
@@ -0,0 +1,106 @@
|
|
|
+import logging
|
|
|
+from datetime import datetime, timedelta
|
|
|
+from pathlib import Path
|
|
|
+from typing import Any, Dict, Optional
|
|
|
+
|
|
|
+import emails
|
|
|
+from emails.template import JinjaTemplate
|
|
|
+from jose import jwt
|
|
|
+
|
|
|
+from core.config import settings
|
|
|
+
|
|
|
+
|
|
|
+def send_email(
|
|
|
+ email_to: str,
|
|
|
+ subject_template: str = "",
|
|
|
+ html_template: str = "",
|
|
|
+ environment: Dict[str, Any] = {},
|
|
|
+) -> None:
|
|
|
+ assert settings.EMAILS_ENABLED, "no provided configuration for email variables"
|
|
|
+ message = emails.Message(
|
|
|
+ subject=JinjaTemplate(subject_template),
|
|
|
+ html=JinjaTemplate(html_template),
|
|
|
+ mail_from=(settings.EMAILS_FROM_NAME, settings.EMAILS_FROM_EMAIL),
|
|
|
+ )
|
|
|
+ smtp_options = {"host": settings.SMTP_HOST, "port": settings.SMTP_PORT}
|
|
|
+ if settings.SMTP_TLS:
|
|
|
+ smtp_options["tls"] = True
|
|
|
+ if settings.SMTP_USER:
|
|
|
+ smtp_options["user"] = settings.SMTP_USER
|
|
|
+ if settings.SMTP_PASSWORD:
|
|
|
+ smtp_options["password"] = settings.SMTP_PASSWORD
|
|
|
+ response = message.send(to=email_to, render=environment, smtp=smtp_options)
|
|
|
+ logging.info(f"send email result: {response}")
|
|
|
+
|
|
|
+
|
|
|
+def send_test_email(email_to: str) -> None:
|
|
|
+ project_name = settings.PROJECT_NAME
|
|
|
+ subject = f"{project_name} - Test email"
|
|
|
+ with open(Path(settings.EMAIL_TEMPLATES_DIR) / "test_email.html") as f:
|
|
|
+ template_str = f.read()
|
|
|
+ send_email(
|
|
|
+ email_to=email_to,
|
|
|
+ subject_template=subject,
|
|
|
+ html_template=template_str,
|
|
|
+ environment={"project_name": settings.PROJECT_NAME, "email": email_to},
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+def send_reset_password_email(email_to: str, email: str, token: str) -> None:
|
|
|
+ project_name = settings.PROJECT_NAME
|
|
|
+ subject = f"{project_name} - Password recovery for user {email}"
|
|
|
+ with open(Path(settings.EMAIL_TEMPLATES_DIR) / "reset_password.html") as f:
|
|
|
+ template_str = f.read()
|
|
|
+ server_host = settings.SERVER_HOST
|
|
|
+ link = f"{server_host}/reset-password?token={token}"
|
|
|
+ send_email(
|
|
|
+ email_to=email_to,
|
|
|
+ subject_template=subject,
|
|
|
+ html_template=template_str,
|
|
|
+ environment={
|
|
|
+ "project_name": settings.PROJECT_NAME,
|
|
|
+ "username": email,
|
|
|
+ "email": email_to,
|
|
|
+ "valid_hours": settings.EMAIL_RESET_TOKEN_EXPIRE_HOURS,
|
|
|
+ "link": link,
|
|
|
+ },
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+def send_new_account_email(email_to: str, username: str, password: str) -> None:
|
|
|
+ project_name = settings.PROJECT_NAME
|
|
|
+ subject = f"{project_name} - New account for user {username}"
|
|
|
+ with open(Path(settings.EMAIL_TEMPLATES_DIR) / "new_account.html") as f:
|
|
|
+ template_str = f.read()
|
|
|
+ link = settings.SERVER_HOST
|
|
|
+ send_email(
|
|
|
+ email_to=email_to,
|
|
|
+ subject_template=subject,
|
|
|
+ html_template=template_str,
|
|
|
+ environment={
|
|
|
+ "project_name": settings.PROJECT_NAME,
|
|
|
+ "username": username,
|
|
|
+ "password": password,
|
|
|
+ "email": email_to,
|
|
|
+ "link": link,
|
|
|
+ },
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+def generate_password_reset_token(email: str) -> str:
|
|
|
+ delta = timedelta(hours=settings.EMAIL_RESET_TOKEN_EXPIRE_HOURS)
|
|
|
+ now = datetime.utcnow()
|
|
|
+ expires = now + delta
|
|
|
+ exp = expires.timestamp()
|
|
|
+ encoded_jwt = jwt.encode(
|
|
|
+ {"exp": exp, "nbf": now, "sub": email}, settings.SECRET_KEY, algorithm="HS256",
|
|
|
+ )
|
|
|
+ return encoded_jwt
|
|
|
+
|
|
|
+
|
|
|
+def verify_password_reset_token(token: str) -> Optional[str]:
|
|
|
+ try:
|
|
|
+ decoded_token = jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"])
|
|
|
+ return decoded_token["email"]
|
|
|
+ except jwt.JWTError:
|
|
|
+ return None
|