95 lines
4.0 KiB
Python
95 lines
4.0 KiB
Python
from fastapi import HTTPException
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
import httpx
|
|
|
|
from notifications.models import EmailConfirmationTokens
|
|
from notifications.plugins import (
|
|
CONFIRM_TOKEN_RESEND_SECONDS,
|
|
CONFIRM_TOKEN_TTL_SECONDS,
|
|
build_confirmation_link,
|
|
compose_token,
|
|
confirmation_expiry,
|
|
generate_token_secret,
|
|
hash_token,
|
|
now_utc,
|
|
render_confirmation_email,
|
|
send_confirmation_mail,
|
|
split_token,
|
|
verify_token,
|
|
)
|
|
from notifications.serializers import serialize_confirmation_request,serialize_confirmation_result
|
|
from users.models import Users
|
|
|
|
|
|
class Confirmation:
|
|
def __init__(self,session:AsyncSession):
|
|
self.session=session
|
|
|
|
async def send_confirmation(self,user):
|
|
"""Issue a fresh token for an already committed Users row and mail the link."""
|
|
await EmailConfirmationTokens.invalidate_tokens_for_user(self.session,str(user.id))
|
|
|
|
secret=generate_token_secret()
|
|
expires_at=confirmation_expiry()
|
|
row=await EmailConfirmationTokens.insert_token(self.session,{
|
|
"user_id":user.id,
|
|
"email":user.email,
|
|
"token_hash":hash_token(secret),
|
|
"expires_at":expires_at,
|
|
})
|
|
|
|
link=build_confirmation_link(compose_token(row.id,secret))
|
|
subject,html=render_confirmation_email(link,CONFIRM_TOKEN_TTL_SECONDS)
|
|
try:
|
|
await send_confirmation_mail(user.email,subject,html)
|
|
except (httpx.HTTPError,RuntimeError) as e:
|
|
await EmailConfirmationTokens.mark_used(self.session,str(row.id))
|
|
raise HTTPException(status_code=502,detail="Failed to send confirmation email") from e
|
|
|
|
return serialize_confirmation_request(user.email,expires_at)
|
|
|
|
async def confirm(self,token):
|
|
record_id,secret=split_token(token)
|
|
if not record_id:
|
|
raise HTTPException(status_code=400,detail="Invalid confirmation link")
|
|
|
|
row=await EmailConfirmationTokens.get_token_by_id(self.session,record_id)
|
|
if not row or not verify_token(secret,row.token_hash):
|
|
raise HTTPException(status_code=400,detail="Invalid confirmation link")
|
|
|
|
user=await Users.get_user_by_id(self.session,str(row.user_id))
|
|
if not user or user.is_deleted:
|
|
raise HTTPException(status_code=404,detail="User not found")
|
|
|
|
# Mail clients, link scanners and the back button all replay this link.
|
|
if row.is_used:
|
|
if row.confirmed_at and user.is_active:
|
|
return serialize_confirmation_result(user,already_confirmed=True)
|
|
raise HTTPException(status_code=400,detail="This confirmation link is no longer valid. Request a new one.")
|
|
|
|
if row.expires_at<=now_utc():
|
|
raise HTTPException(status_code=400,detail="Confirmation link has expired. Request a new one.")
|
|
|
|
if user.is_active:
|
|
await EmailConfirmationTokens.mark_confirmed(self.session,str(row.id))
|
|
return serialize_confirmation_result(user,already_confirmed=True)
|
|
|
|
updated=await Users.update_user(self.session,str(user.id),{"is_active":True})
|
|
await EmailConfirmationTokens.mark_confirmed(self.session,str(row.id))
|
|
return serialize_confirmation_result(updated)
|
|
|
|
async def resend(self,email):
|
|
user=await Users.get_user_by_email(self.session,email)
|
|
if not user or user.is_deleted:
|
|
raise HTTPException(status_code=404,detail="No account found for this email")
|
|
if user.is_active:
|
|
raise HTTPException(status_code=400,detail="This account is already confirmed")
|
|
|
|
active=await EmailConfirmationTokens.get_active_token_by_user(self.session,str(user.id))
|
|
if active:
|
|
age=(now_utc()-active.created_at).total_seconds()
|
|
if age<CONFIRM_TOKEN_RESEND_SECONDS and active.expires_at>now_utc():
|
|
raise HTTPException(status_code=429,detail="Please wait before requesting another confirmation email")
|
|
|
|
return await self.send_confirmation(user)
|