import os
import smtplib
import ssl
import time
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from flask import Blueprint, flash, jsonify, redirect, request, url_for
from app.models import (
generate_password_reset_token,
generate_verification_token,
reset_password_with_token,
verify_email_token,
)
from app.routes.auth import current_user, login_required
from app.services.logging import log
email_bp = Blueprint("email", __name__)
# SMTP config from env
SMTP_HOST = os.environ.get("SMTP_HOST", "")
SMTP_PORT = int(os.environ.get("SMTP_PORT", "587"))
SMTP_USER = os.environ.get("SMTP_USER", "")
SMTP_PASS = os.environ.get("SMTP_PASS", "")
SMTP_FROM = os.environ.get("SMTP_FROM", "noreply@agentforms.io")
class NotificationService:
"""SMTP email service for sending notifications with circuit breaker."""
# Circuit breaker state
_smtp_failure_count = 0
_smtp_last_failure_time = 0
_smtp_circuit_open = False
_SMTP_MAX_FAILURES = 5
_SMTP_INITIAL_BACKOFF = 60 # seconds
_SMTP_MAX_BACKOFF = 300 # 5 minutes
@staticmethod
def _check_circuit_breaker():
"""Check if circuit breaker is open. Returns True if circuit is open (should not send)."""
if not NotificationService._smtp_circuit_open:
return False
# Check if backoff period has elapsed
elapsed = time.time() - NotificationService._smtp_last_failure_time
backoff = min(
NotificationService._SMTP_INITIAL_BACKOFF
* (2 ** (NotificationService._smtp_failure_count - NotificationService._SMTP_MAX_FAILURES)),
NotificationService._SMTP_MAX_BACKOFF,
)
if elapsed >= backoff:
# Half-open: allow one attempt
return False
return True
@staticmethod
def _record_success():
"""Reset circuit breaker on success."""
NotificationService._smtp_failure_count = 0
NotificationService._smtp_circuit_open = False
@staticmethod
def _record_failure():
"""Record a failure and potentially open the circuit."""
NotificationService._smtp_failure_count += 1
NotificationService._smtp_last_failure_time = time.time()
if NotificationService._smtp_failure_count >= NotificationService._SMTP_MAX_FAILURES:
NotificationService._smtp_circuit_open = True
log.warning("smtp", f"Circuit breaker OPEN after {NotificationService._smtp_failure_count} failures")
@staticmethod
def circuit_status():
"""Return circuit breaker status."""
return {
"open": NotificationService._smtp_circuit_open,
"failure_count": NotificationService._smtp_failure_count,
"last_failure_time": NotificationService._smtp_last_failure_time,
}
@staticmethod
def is_configured():
"""Check if SMTP is properly configured."""
return bool(SMTP_HOST)
@staticmethod
def status():
"""Return SMTP configuration status (no connection test)."""
return {
"configured": bool(SMTP_HOST),
"host": SMTP_HOST or None,
"port": SMTP_PORT if SMTP_HOST else None,
"username": SMTP_USER or None,
"from": SMTP_FROM or None,
"tls": bool(SMTP_PORT in (587, 465)),
"circuit_breaker": NotificationService.circuit_status(),
}
@staticmethod
def send(to: str, subject: str, body: str, from_addr: str = None) -> bool:
"""Send a plain-text email via configured SMTP.
Args:
to: Recipient address
subject: Email subject
body: Plain-text body
from_addr: Override SMTP_FROM (optional)
Returns:
True if sent successfully, False otherwise
"""
if not SMTP_HOST:
log.warning("smtp", "Not configured, skipping")
return False
if NotificationService._check_circuit_breaker():
log.warning("smtp", "Circuit breaker open, skipping send")
return False
msg = MIMEText(body, "plain")
msg["From"] = from_addr or SMTP_FROM
msg["To"] = to
msg["Subject"] = subject
try:
context = ssl.create_default_context()
with smtplib.SMTP(SMTP_HOST, SMTP_PORT, timeout=15) as server:
server.ehlo()
if SMTP_USER and SMTP_PASS:
server.starttls(context=context)
server.ehlo()
server.login(SMTP_USER, SMTP_PASS)
server.send_message(msg)
NotificationService._record_success()
log.info("smtp", f"Sent to {to}")
return True
except Exception as e:
NotificationService._record_failure()
log.error("smtp", "Failed to send email", error=str(e))
return False
@staticmethod
def send_html(to: str, subject: str, html_body: str, plain_body: str = None, from_addr: str = None) -> bool:
"""Send an HTML email with optional plain-text fallback via configured SMTP.
Args:
to: Recipient address
subject: Email subject
html_body: HTML body
plain_body: Plain-text fallback (optional)
from_addr: Override SMTP_FROM (optional)
Returns:
True if sent successfully, False otherwise
"""
if not SMTP_HOST:
log.warning("smtp", "Not configured, skipping")
return False
if NotificationService._check_circuit_breaker():
log.warning("smtp", "Circuit breaker open, skipping send")
return False
msg = MIMEMultipart("alternative")
msg["From"] = from_addr or SMTP_FROM
msg["To"] = to
msg["Subject"] = subject
if plain_body:
msg.attach(MIMEText(plain_body, "plain"))
msg.attach(MIMEText(html_body, "html"))
try:
context = ssl.create_default_context()
with smtplib.SMTP(SMTP_HOST, SMTP_PORT, timeout=15) as server:
server.ehlo()
if SMTP_USER and SMTP_PASS:
server.starttls(context=context)
server.ehlo()
server.login(SMTP_USER, SMTP_PASS)
server.send_message(msg)
NotificationService._record_success()
log.info("smtp", f"Sent HTML to {to}")
return True
except Exception as e:
NotificationService._record_failure()
log.error("smtp", "Failed to send HTML email", error=str(e))
return False
@email_bp.route("/api/email/smtp_status")
def smtp_status():
"""GET — Return SMTP configuration status."""
return jsonify(NotificationService.status())
@email_bp.route("/api/email/send", methods=["POST"])
@login_required
def send_email():
"""POST — Send an email via SMTP.
JSON body:
to: recipient email
subject: email subject
body: plain-text body
"""
data = request.get_json(silent=True) or {}
to = data.get("to", "").strip()
subject = data.get("subject", "").strip()
body = data.get("body", "").strip()
if not to or not subject or not body:
return jsonify({"error": "Missing required fields: to, subject, body"}), 400
success = NotificationService.send(to, subject, body)
if success:
return jsonify({"success": True, "to": to})
else:
return jsonify({"error": "Failed to send email. Check SMTP configuration."}), 502
# ─── Email Verification ─────────────────────────────────────────────────────
@email_bp.route("/email/verify/send", methods=["GET", "POST"])
@login_required
def send_verification():
"""Send a verification email to the current user."""
user = current_user()
token = generate_verification_token(user["id"])
# Build verification link
base_url = request.host_url.rstrip("/")
verify_link = f"{base_url}/email/verify/{token}"
body = f"""Hello {user.get("name", "there")},
Please verify your email address by clicking the link below:
{verify_link}
This link expires in 24 hours.
If you didn't request this, you can ignore this email.
— AgentForms
"""
success = NotificationService.send(user["email"], "Verify your email — AgentForms", body)
if request.is_json or request.method == "POST":
if success:
return jsonify({"success": True, "message": "Verification email sent"})
else:
return jsonify({"error": "Failed to send verification email"}), 502
# GET redirect (from verified_email_required decorator)
flash("We've sent a verification email to your address.", "info")
return redirect(url_for("auth.dashboard"))
@email_bp.route("/email/verify/<token>")
def verify_email(token):
"""Verify email via token link. Returns a page or JSON."""
user_id = verify_email_token(token)
if user_id:
if request.is_json or request.accept_mimetypes.best == "application/json":
return jsonify({"success": True, "message": "Email verified"})
return redirect(url_for("auth.login", message="Email verified successfully"))
if request.is_json or request.accept_mimetypes.best == "application/json":
return jsonify({"error": "Invalid or expired verification token"}), 400
return redirect(url_for("auth.login", message="Invalid or expired verification link"))
# ─── Password Reset ─────────────────────────────────────────────────────────
@email_bp.route("/email/reset/request", methods=["POST"])
def request_password_reset():
"""Request a password reset email.
JSON body:
email: user email
"""
data = request.get_json(silent=True) or {}
email = data.get("email", "").strip()
if not email or "@" not in email:
return jsonify({"error": "Valid email is required"}), 400
token = generate_password_reset_token(email)
if token:
base_url = request.host_url.rstrip("/")
reset_link = f"{base_url}/email/reset/{token}"
body = f"""Hello,
You requested a password reset for your AgentForms account.
Click the link below to reset your password:
{reset_link}
This link expires in 1 hour.
If you didn't request this, you can ignore this email. Your password will not be changed.
— AgentForms
"""
# Find the user to send email
from app.crypto import hash_value, try_decrypt_user_value
from app.models import get_db
conn = None
try:
conn = get_db()
user = conn.execute(
"SELECT id, email, email_encrypted FROM users WHERE email_hash = ?", (hash_value(email),)
).fetchone()
finally:
if conn:
conn.close()
if user:
decrypted = try_decrypt_user_value(user["id"], user["email_encrypted"]) or user["email"]
NotificationService.send(decrypted, "Reset your password — AgentForms", body)
# Always return success to avoid email enumeration
return jsonify({"success": True, "message": "If an account with that email exists, you'll receive a reset link."})
@email_bp.route("/email/reset/<token>", methods=["POST"])
def reset_password(token):
"""Reset password using token.
JSON body:
password: new password
"""
data = request.get_json(silent=True) or {}
password = data.get("password", "").strip()
if not password or len(password) < 8:
return jsonify({"error": "Password must be at least 8 characters"}), 400
user_id = reset_password_with_token(token, password)
if user_id:
return jsonify({"success": True, "message": "Password reset successfully"})
else:
return jsonify({"error": "Invalid or expired reset token"}), 400