import logging from datetime import datetime, timedelta from pathlib import Path from typing import Optional import emails import jwt from emails.template import JinjaTemplate from jwt.exceptions import InvalidTokenError from app.core import config password_reset_jwt_subject = "preset" def send_email(email_to: str, subject_template="", html_template="", environment={}): assert config.EMAILS_ENABLED, "no provided configuration for email variables" message = emails.Message( subject=JinjaTemplate(subject_template), html=JinjaTemplate(html_template), mail_from=(config.EMAILS_FROM_NAME, config.EMAILS_FROM_EMAIL), ) smtp_options = {"host": config.SMTP_HOST, "port": config.SMTP_PORT} if config.SMTP_TLS: smtp_options["tls"] = True if config.SMTP_USER: smtp_options["user"] = config.SMTP_USER if config.SMTP_PASSWORD: smtp_options["password"] = config.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): project_name = config.PROJECT_NAME subject = f"{project_name} - Test email" with open(Path(config.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": config.PROJECT_NAME, "email": email_to}, ) def send_reset_password_email(email_to: str, email: str, token: str): project_name = config.PROJECT_NAME subject = f"{project_name} - Password recovery for user {email}" with open(Path(config.EMAIL_TEMPLATES_DIR) / "reset_password.html") as f: template_str = f.read() if hasattr(token, "decode"): use_token = token.decode() else: use_token = token server_host = config.SERVER_HOST link = f"{server_host}/reset-password?token={use_token}" send_email( email_to=email_to, subject_template=subject, html_template=template_str, environment={ "project_name": config.PROJECT_NAME, "username": email, "email": email_to, "valid_hours": config.EMAIL_RESET_TOKEN_EXPIRE_HOURS, "link": link, }, ) def send_new_account_email(email_to: str, username: str, password: str): project_name = config.PROJECT_NAME subject = f"{project_name} - New account for user {username}" with open(Path(config.EMAIL_TEMPLATES_DIR) / "new_account.html") as f: template_str = f.read() link = config.SERVER_HOST send_email( email_to=email_to, subject_template=subject, html_template=template_str, environment={ "project_name": config.PROJECT_NAME, "username": username, "password": password, "email": email_to, "link": link, }, ) def generate_password_reset_token(email): delta = timedelta(hours=config.EMAIL_RESET_TOKEN_EXPIRE_HOURS) now = datetime.utcnow() expires = now + delta exp = expires.timestamp() encoded_jwt = jwt.encode( {"exp": exp, "nbf": now, "sub": password_reset_jwt_subject, "email": email}, config.SECRET_KEY, algorithm="HS256", ) return encoded_jwt def verify_password_reset_token(token) -> Optional[str]: try: decoded_token = jwt.decode(token, config.SECRET_KEY, algorithms=["HS256"]) assert decoded_token["sub"] == password_reset_jwt_subject return decoded_token["email"] except InvalidTokenError: return None