cd112bd683
* ♻️ Refactor and simplify backend code * ♻️ Refactor frontend state, integrate typesafe-vuex accessors into state files * ♻️ Use new state accessors and standardize layout * 🔒 Upgrade and fix npm security audit * 🔧 Update local re-generation scripts * 🔊 Log startup exceptions to detect errors early * ✏️ Fix password reset token content * 🔥 Remove unneeded Dockerfile directives * 🔥 Remove unnecessary print * 🔥 Remove unnecessary code, upgrade dependencies in backend * ✏️ Fix typos in docstrings and comments * 🏗️ Improve user Depends utilities to simplify and remove code * 🔥 Remove deprecated SQLAlchemy parameter
112 lines
3.6 KiB
Python
112 lines
3.6 KiB
Python
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
|