116 lines
3.8 KiB
Python
116 lines
3.8 KiB
Python
"""Confirmation helpers — token generation, hashing, link building, and Teams mail send.
|
|
|
|
Pure module: no FastAPI imports and no HTTPException.
|
|
|
|
`send_confirmation_mail` intentionally duplicates
|
|
`forget_password.plugins.send_reset_mail` rather than importing it: that helper is
|
|
domain-named, and each domain owns its own mail copy and its own env reads.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import secrets
|
|
import uuid
|
|
from datetime import datetime, timedelta, timezone
|
|
from urllib.parse import quote
|
|
|
|
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")
|
|
FRONTEND_URL = os.getenv("FRONTEND_URL", "http://localhost:5173")
|
|
CONFIRM_EMAIL_PATH = os.getenv("CONFIRM_EMAIL_PATH", "/auth/confirm-email")
|
|
CONFIRM_TOKEN_TTL_SECONDS = int(os.getenv("CONFIRM_TOKEN_TTL_SECONDS", "86400"))
|
|
CONFIRM_TOKEN_RESEND_SECONDS = int(os.getenv("CONFIRM_TOKEN_RESEND_SECONDS", "60"))
|
|
MAIL_ACCEPTED_STATUS = 202
|
|
|
|
# 32 bytes -> 43 url-safe characters, well under users.plugins.BCRYPT_MAX_BYTES.
|
|
TOKEN_SECRET_BYTES = 32
|
|
|
|
|
|
def now_utc() -> datetime:
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
def confirmation_expiry(*, now: datetime | None = None) -> datetime:
|
|
return (now or now_utc()) + timedelta(seconds=CONFIRM_TOKEN_TTL_SECONDS)
|
|
|
|
|
|
def generate_token_secret() -> str:
|
|
return secrets.token_urlsafe(TOKEN_SECRET_BYTES)
|
|
|
|
|
|
def hash_token(secret: str) -> str:
|
|
return hash_password(secret)
|
|
|
|
|
|
def verify_token(secret: str, token_hash: str) -> bool:
|
|
return verify_password(secret, token_hash)
|
|
|
|
|
|
def compose_token(record_id, secret: str) -> str:
|
|
"""Link token. A bcrypt hash cannot be looked up, so the row id rides along."""
|
|
return f"{record_id}.{secret}"
|
|
|
|
|
|
def split_token(token: str) -> tuple[str | None, str | None]:
|
|
"""('<uuid>','<secret>') or (None,None). token_urlsafe never emits a dot."""
|
|
if not token or "." not in token:
|
|
return None, None
|
|
record_id, secret = token.split(".", 1)
|
|
if not record_id or not secret:
|
|
return None, None
|
|
try:
|
|
uuid.UUID(record_id)
|
|
except ValueError:
|
|
return None, None
|
|
return record_id, secret
|
|
|
|
|
|
def build_confirmation_link(token: str) -> str:
|
|
base = FRONTEND_URL.rstrip("/")
|
|
path = CONFIRM_EMAIL_PATH if CONFIRM_EMAIL_PATH.startswith("/") else f"/{CONFIRM_EMAIL_PATH}"
|
|
return f"{base}{path}?token={quote(token, safe='')}"
|
|
|
|
|
|
def render_confirmation_email(link: str, ttl: int) -> tuple[str, str]:
|
|
hours = max(1, ttl // 3600)
|
|
subject = "Confirm your TalentFlow account"
|
|
html = (
|
|
"<p>Welcome to TalentFlow. Confirm your email address to activate your account.</p>"
|
|
f'<p><a href="{link}">Confirm my email</a></p>'
|
|
f"<p>This link expires in {hours} hour(s). If you did not sign up, ignore this email.</p>"
|
|
f"<p>If the link does not open, paste this into your browser:<br>{link}</p>"
|
|
)
|
|
return subject, html
|
|
|
|
|
|
async def send_confirmation_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,
|
|
)
|