refactor: Use SES v2 stored templates in Python email service
- Switch from raw HTML to stored templates - Use sesv2 client instead of ses - Add send_security_alert_email function - Templates: ELMEG_EMAIL_VERIFICATION, ELMEG_PASSWORD_RESET, ELMEG_SECURITY_ALERT
This commit is contained in:
fullsizemalt
parent
530f217445
commit
415a092257
1 changed files with 109 additions and 78 deletions
|
|
@ -1,115 +1,146 @@
|
|||
"""
|
||||
Email Service - AWS SES integration for verification and password reset emails.
|
||||
Email Service - AWS SES v2 integration using stored templates.
|
||||
|
||||
Uses SES stored templates for consistent, branded transactional emails:
|
||||
- ELMEG_EMAIL_VERIFICATION
|
||||
- ELMEG_PASSWORD_RESET
|
||||
- ELMEG_SECURITY_ALERT
|
||||
"""
|
||||
import os
|
||||
import json
|
||||
import secrets
|
||||
from datetime import datetime, timedelta
|
||||
from typing import Optional
|
||||
import boto3
|
||||
from botocore.exceptions import ClientError
|
||||
|
||||
# Configuration
|
||||
AWS_REGION = os.getenv("AWS_SES_REGION", "us-east-1")
|
||||
EMAIL_FROM = os.getenv("EMAIL_FROM", "noreply@elmeg.xyz")
|
||||
FRONTEND_URL = os.getenv("FRONTEND_URL", "https://elmeg.xyz")
|
||||
SUPPORT_EMAIL = os.getenv("SUPPORT_EMAIL", "support@elmeg.xyz")
|
||||
APP_NAME = "Elmeg"
|
||||
|
||||
# SES Template Names
|
||||
TEMPLATE_VERIFICATION = "ELMEG_EMAIL_VERIFICATION"
|
||||
TEMPLATE_PASSWORD_RESET = "ELMEG_PASSWORD_RESET"
|
||||
TEMPLATE_SECURITY_ALERT = "ELMEG_SECURITY_ALERT"
|
||||
|
||||
# AWS credentials from environment (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY)
|
||||
# or from IAM role if running on AWS infrastructure
|
||||
|
||||
def get_ses_client():
|
||||
"""Get boto3 SES client"""
|
||||
return boto3.client('ses', region_name=AWS_REGION)
|
||||
"""Get boto3 SES v2 client"""
|
||||
return boto3.client('sesv2', region_name=AWS_REGION)
|
||||
|
||||
|
||||
def send_email(to: str, subject: str, html_content: str) -> bool:
|
||||
"""Send email via AWS SES"""
|
||||
# Dev mode - just log
|
||||
if not os.getenv("AWS_ACCESS_KEY_ID"):
|
||||
print(f"[EMAIL DEV MODE] To: {to}, Subject: {subject}")
|
||||
print(f"[EMAIL DEV MODE] Content: {html_content[:200]}...")
|
||||
return True
|
||||
def is_email_configured() -> bool:
|
||||
"""Check if email is properly configured"""
|
||||
return bool(os.getenv("AWS_ACCESS_KEY_ID") and os.getenv("AWS_SECRET_ACCESS_KEY"))
|
||||
|
||||
|
||||
def send_templated_email(
|
||||
to: str,
|
||||
template_name: str,
|
||||
template_data: dict
|
||||
) -> dict:
|
||||
"""
|
||||
Send email using SES stored template.
|
||||
|
||||
Returns:
|
||||
dict with 'success', 'message_id' (on success), 'error' (on failure)
|
||||
"""
|
||||
# Dev mode - log instead of sending
|
||||
if not is_email_configured():
|
||||
print(f"[EMAIL DEV MODE] To: {to}, Template: {template_name}")
|
||||
print(f"[EMAIL DEV MODE] Data: {json.dumps(template_data, indent=2)}")
|
||||
return {"success": True, "message_id": "dev-mode", "dev_mode": True}
|
||||
|
||||
try:
|
||||
client = get_ses_client()
|
||||
response = client.send_email(
|
||||
Source=EMAIL_FROM,
|
||||
Destination={'ToAddresses': [to]},
|
||||
Message={
|
||||
'Subject': {'Data': subject, 'Charset': 'UTF-8'},
|
||||
'Body': {
|
||||
'Html': {'Data': html_content, 'Charset': 'UTF-8'}
|
||||
FromEmailAddress=EMAIL_FROM,
|
||||
Destination={"ToAddresses": [to]},
|
||||
Content={
|
||||
"Template": {
|
||||
"TemplateName": template_name,
|
||||
"TemplateData": json.dumps(template_data)
|
||||
}
|
||||
}
|
||||
)
|
||||
print(f"Email sent to {to}, MessageId: {response['MessageId']}")
|
||||
return True
|
||||
message_id = response.get("MessageId", "unknown")
|
||||
print(f"[Email] Sent {template_name} to {to}, MessageId: {message_id}")
|
||||
return {"success": True, "message_id": message_id}
|
||||
|
||||
except ClientError as e:
|
||||
print(f"Failed to send email: {e.response['Error']['Message']}")
|
||||
return False
|
||||
error_msg = e.response.get('Error', {}).get('Message', str(e))
|
||||
print(f"[Email] Failed to send {template_name} to {to}: {error_msg}")
|
||||
return {"success": False, "error": error_msg}
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Email Functions
|
||||
# =============================================================================
|
||||
|
||||
async def send_verification_email(email: str, token: str, user_name: Optional[str] = None) -> bool:
|
||||
"""Send email verification using SES template"""
|
||||
verification_link = f"{FRONTEND_URL}/verify-email?token={token}"
|
||||
|
||||
template_data = {
|
||||
"user_name": user_name or email.split("@")[0],
|
||||
"verification_link": verification_link,
|
||||
"app_name": APP_NAME,
|
||||
"support_email": SUPPORT_EMAIL
|
||||
}
|
||||
|
||||
result = send_templated_email(email, TEMPLATE_VERIFICATION, template_data)
|
||||
return result["success"]
|
||||
|
||||
|
||||
async def send_password_reset_email(email: str, token: str, user_name: Optional[str] = None) -> bool:
|
||||
"""Send password reset email using SES template"""
|
||||
reset_link = f"{FRONTEND_URL}/reset-password?token={token}"
|
||||
|
||||
template_data = {
|
||||
"user_name": user_name or email.split("@")[0],
|
||||
"reset_link": reset_link,
|
||||
"app_name": APP_NAME,
|
||||
"support_email": SUPPORT_EMAIL
|
||||
}
|
||||
|
||||
result = send_templated_email(email, TEMPLATE_PASSWORD_RESET, template_data)
|
||||
return result["success"]
|
||||
|
||||
|
||||
async def send_security_alert_email(
|
||||
email: str,
|
||||
security_event_description: str,
|
||||
user_name: Optional[str] = None
|
||||
) -> bool:
|
||||
"""Send security alert email using SES template"""
|
||||
template_data = {
|
||||
"user_name": user_name or email.split("@")[0],
|
||||
"security_event_description": security_event_description,
|
||||
"app_name": APP_NAME,
|
||||
"support_email": SUPPORT_EMAIL
|
||||
}
|
||||
|
||||
result = send_templated_email(email, TEMPLATE_SECURITY_ALERT, template_data)
|
||||
return result["success"]
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Token Generation & Expiry Helpers
|
||||
# =============================================================================
|
||||
|
||||
def generate_token() -> str:
|
||||
"""Generate a secure random token"""
|
||||
return secrets.token_urlsafe(32)
|
||||
|
||||
|
||||
async def send_verification_email(email: str, token: str) -> bool:
|
||||
"""Send email verification link"""
|
||||
verify_url = f"{FRONTEND_URL}/verify-email?token={token}"
|
||||
|
||||
html = f"""
|
||||
<div style="font-family: sans-serif; max-width: 600px; margin: 0 auto;">
|
||||
<h1 style="color: #333;">Welcome to Elmeg!</h1>
|
||||
<p>Please verify your email address by clicking the button below:</p>
|
||||
<a href="{verify_url}"
|
||||
style="display: inline-block; background: #4F46E5; color: white;
|
||||
padding: 12px 24px; text-decoration: none; border-radius: 6px;
|
||||
margin: 16px 0;">
|
||||
Verify Email
|
||||
</a>
|
||||
<p style="color: #666; font-size: 14px;">
|
||||
Or copy this link: {verify_url}
|
||||
</p>
|
||||
<p style="color: #999; font-size: 12px;">
|
||||
This link expires in 48 hours.
|
||||
</p>
|
||||
</div>
|
||||
"""
|
||||
|
||||
return send_email(email, "Verify your Elmeg account", html)
|
||||
|
||||
|
||||
async def send_password_reset_email(email: str, token: str) -> bool:
|
||||
"""Send password reset link"""
|
||||
reset_url = f"{FRONTEND_URL}/reset-password?token={token}"
|
||||
|
||||
html = f"""
|
||||
<div style="font-family: sans-serif; max-width: 600px; margin: 0 auto;">
|
||||
<h1 style="color: #333;">Password Reset</h1>
|
||||
<p>You requested a password reset. Click below to set a new password:</p>
|
||||
<a href="{reset_url}"
|
||||
style="display: inline-block; background: #4F46E5; color: white;
|
||||
padding: 12px 24px; text-decoration: none; border-radius: 6px;
|
||||
margin: 16px 0;">
|
||||
Reset Password
|
||||
</a>
|
||||
<p style="color: #666; font-size: 14px;">
|
||||
Or copy this link: {reset_url}
|
||||
</p>
|
||||
<p style="color: #999; font-size: 12px;">
|
||||
This link expires in 1 hour. If you didn't request this, ignore this email.
|
||||
</p>
|
||||
</div>
|
||||
"""
|
||||
|
||||
return send_email(email, "Reset your Elmeg password", html)
|
||||
|
||||
|
||||
# Token expiration helpers
|
||||
def get_verification_expiry() -> datetime:
|
||||
"""48 hour expiry for email verification"""
|
||||
return datetime.utcnow() + timedelta(hours=48)
|
||||
"""24 hour expiry for email verification"""
|
||||
return datetime.utcnow() + timedelta(hours=24)
|
||||
|
||||
|
||||
def get_reset_expiry() -> datetime:
|
||||
"""1 hour expiry for password reset"""
|
||||
return datetime.utcnow() + timedelta(hours=1)
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue