78 lines
2.3 KiB
Python
78 lines
2.3 KiB
Python
"""Forget-password helpers — OTP generation, hashing, and Teams mail send.
|
|
|
|
Pure module: no FastAPI imports and no HTTPException.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import secrets
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import httpx
|
|
from dotenv import load_dotenv
|
|
|
|
from users.plugins import hash_password, verify_password
|
|
|
|
load_dotenv()
|
|
|
|
TEAMS_MAIL_API_URL = os.getenv("TEAMS_MAIL_API_URL")
|
|
TEAMS_API_TOKEN = os.getenv("TEAMS_API_TOKEN")
|
|
RESET_CODE_TTL_SECONDS = int(os.getenv("RESET_CODE_TTL_SECONDS", "60"))
|
|
RESET_CODE_RESEND_SECONDS = int(os.getenv("RESET_CODE_RESEND_SECONDS", "30"))
|
|
RESET_CODE_MAX_ATTEMPTS = int(os.getenv("RESET_CODE_MAX_ATTEMPTS", "5"))
|
|
MAIL_ACCEPTED_STATUS = 202
|
|
|
|
|
|
def now_utc() -> datetime:
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
def code_expiry(*, now: datetime | None = None) -> datetime:
|
|
return (now or now_utc()) + timedelta(seconds=RESET_CODE_TTL_SECONDS)
|
|
|
|
|
|
def generate_code() -> str:
|
|
return f"{secrets.randbelow(1_000_000):06d}"
|
|
|
|
|
|
def hash_code(code: str) -> str:
|
|
return hash_password(code)
|
|
|
|
|
|
def verify_code(code: str, code_hash: str) -> bool:
|
|
return verify_password(code, code_hash)
|
|
|
|
|
|
def render_reset_email(code: str, ttl: int) -> tuple[str, str]:
|
|
subject = "Your TalentFlow password reset code"
|
|
html = (
|
|
f"<p>Your password reset code is <strong>{code}</strong>.</p>"
|
|
f"<p>It expires in {ttl} seconds. If you did not request this, ignore this email.</p>"
|
|
)
|
|
return subject, html
|
|
|
|
|
|
async def send_reset_mail(to_email: str, subject: str, html: str) -> None:
|
|
if not TEAMS_MAIL_API_URL or not TEAMS_API_TOKEN:
|
|
raise RuntimeError("TEAMS_MAIL_API_URL and TEAMS_API_TOKEN must be set")
|
|
fields = [
|
|
("subject", (None, subject)),
|
|
("body", (None, html)),
|
|
("content_type", (None, "html")),
|
|
("save_to_sent_items", (None, "false")),
|
|
("to", (None, to_email)),
|
|
]
|
|
async with httpx.AsyncClient(timeout=15.0) as client:
|
|
response = await client.post(
|
|
TEAMS_MAIL_API_URL,
|
|
files=fields,
|
|
headers={"Authorization": f"Bearer {TEAMS_API_TOKEN}"},
|
|
)
|
|
if response.status_code != MAIL_ACCEPTED_STATUS:
|
|
raise httpx.HTTPStatusError(
|
|
response.text,
|
|
request=response.request,
|
|
response=response,
|
|
)
|