1f0dcb1ddb
- Frontend: Next.js 14 with TypeScript - Backend: FastAPI with SQLAlchemy - Agent: Carmodoo sync agent - Deployment: Docker Compose based staging/production setup - Scripts: Automated deployment with rollback support 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
314 lines
9.7 KiB
Python
314 lines
9.7 KiB
Python
"""
|
||
Verification Service for Email and SMS
|
||
Handles sending and verifying codes for user authentication
|
||
"""
|
||
import random
|
||
import string
|
||
import smtplib
|
||
from email.mime.text import MIMEText
|
||
from email.mime.multipart import MIMEMultipart
|
||
from datetime import datetime, timedelta
|
||
from typing import Optional, Tuple
|
||
from sqlalchemy.orm import Session
|
||
|
||
from ..config import get_settings
|
||
from ..models.user import User, VerificationCode
|
||
|
||
settings = get_settings()
|
||
|
||
|
||
def generate_code(length: int = 6) -> str:
|
||
"""Generate a random numeric code"""
|
||
return ''.join(random.choices(string.digits, k=length))
|
||
|
||
|
||
def create_verification_code(
|
||
db: Session,
|
||
code_type: str, # 'email' or 'phone'
|
||
email: Optional[str] = None,
|
||
phone: Optional[str] = None,
|
||
user_id: Optional[int] = None,
|
||
purpose: str = "verification"
|
||
) -> VerificationCode:
|
||
"""Create a new verification code"""
|
||
# Invalidate any existing codes for this email/phone
|
||
if email:
|
||
db.query(VerificationCode).filter(
|
||
VerificationCode.email == email,
|
||
VerificationCode.code_type == code_type,
|
||
VerificationCode.verified_at.is_(None)
|
||
).delete()
|
||
if phone:
|
||
db.query(VerificationCode).filter(
|
||
VerificationCode.phone == phone,
|
||
VerificationCode.code_type == code_type,
|
||
VerificationCode.verified_at.is_(None)
|
||
).delete()
|
||
|
||
# Create new code
|
||
code = VerificationCode(
|
||
user_id=user_id,
|
||
email=email,
|
||
phone=phone,
|
||
code=generate_code(),
|
||
code_type=code_type,
|
||
purpose=purpose,
|
||
expires_at=datetime.utcnow() + timedelta(minutes=settings.VERIFICATION_CODE_EXPIRE_MINUTES)
|
||
)
|
||
db.add(code)
|
||
db.commit()
|
||
db.refresh(code)
|
||
return code
|
||
|
||
|
||
def verify_code(
|
||
db: Session,
|
||
code: str,
|
||
code_type: str,
|
||
email: Optional[str] = None,
|
||
phone: Optional[str] = None
|
||
) -> Tuple[bool, str]:
|
||
"""
|
||
Verify a code and return (success, message)
|
||
"""
|
||
query = db.query(VerificationCode).filter(
|
||
VerificationCode.code_type == code_type,
|
||
VerificationCode.verified_at.is_(None)
|
||
)
|
||
|
||
if email:
|
||
query = query.filter(VerificationCode.email == email)
|
||
if phone:
|
||
query = query.filter(VerificationCode.phone == phone)
|
||
|
||
verification = query.order_by(VerificationCode.created_at.desc()).first()
|
||
|
||
if not verification:
|
||
return False, "No verification code found. Please request a new one."
|
||
|
||
# Check if expired
|
||
if datetime.utcnow() > verification.expires_at.replace(tzinfo=None):
|
||
return False, "Verification code has expired. Please request a new one."
|
||
|
||
# Check attempts
|
||
if verification.attempts >= verification.max_attempts:
|
||
return False, "Too many failed attempts. Please request a new code."
|
||
|
||
# Check code
|
||
if verification.code != code:
|
||
verification.attempts += 1
|
||
db.commit()
|
||
remaining = verification.max_attempts - verification.attempts
|
||
return False, f"Invalid code. {remaining} attempts remaining."
|
||
|
||
# Success
|
||
verification.verified_at = datetime.utcnow()
|
||
db.commit()
|
||
|
||
return True, "Verification successful"
|
||
|
||
|
||
async def send_email_verification(
|
||
db: Session,
|
||
email: str,
|
||
user_id: Optional[int] = None,
|
||
language: str = "en"
|
||
) -> Tuple[bool, str]:
|
||
"""Send email verification code"""
|
||
# Check rate limit (1 email per minute)
|
||
recent = db.query(VerificationCode).filter(
|
||
VerificationCode.email == email,
|
||
VerificationCode.code_type == "email",
|
||
VerificationCode.created_at > datetime.utcnow() - timedelta(minutes=1)
|
||
).first()
|
||
|
||
if recent:
|
||
return False, "Please wait 1 minute before requesting another code."
|
||
|
||
# Create verification code
|
||
verification = create_verification_code(
|
||
db=db,
|
||
code_type="email",
|
||
email=email,
|
||
user_id=user_id
|
||
)
|
||
|
||
# Send email
|
||
try:
|
||
# Email templates by language
|
||
subjects = {
|
||
"en": "AutonetSellCar - Email Verification Code",
|
||
"ko": "AutonetSellCar - 이메일 인증 코드",
|
||
"mn": "AutonetSellCar - Имэйл баталгаажуулах код",
|
||
"ru": "AutonetSellCar - Код подтверждения email"
|
||
}
|
||
|
||
bodies = {
|
||
"en": f"""
|
||
Hello,
|
||
|
||
Your verification code is: {verification.code}
|
||
|
||
This code will expire in {settings.VERIFICATION_CODE_EXPIRE_MINUTES} minutes.
|
||
|
||
If you didn't request this code, please ignore this email.
|
||
|
||
Best regards,
|
||
AutonetSellCar Team
|
||
""",
|
||
"ko": f"""
|
||
안녕하세요,
|
||
|
||
인증 코드: {verification.code}
|
||
|
||
이 코드는 {settings.VERIFICATION_CODE_EXPIRE_MINUTES}분 후에 만료됩니다.
|
||
|
||
요청하지 않은 경우 이 이메일을 무시하세요.
|
||
|
||
감사합니다,
|
||
AutonetSellCar 팀
|
||
""",
|
||
"mn": f"""
|
||
Сайн байна уу,
|
||
|
||
Таны баталгаажуулах код: {verification.code}
|
||
|
||
Энэ код {settings.VERIFICATION_CODE_EXPIRE_MINUTES} минутын дараа хүчингүй болно.
|
||
|
||
Хэрэв та энэ кодыг хүсээгүй бол энэ имэйлийг үл тоомсорлоно уу.
|
||
|
||
Хүндэтгэсэн,
|
||
AutonetSellCar баг
|
||
""",
|
||
"ru": f"""
|
||
Здравствуйте,
|
||
|
||
Ваш код подтверждения: {verification.code}
|
||
|
||
Этот код истечет через {settings.VERIFICATION_CODE_EXPIRE_MINUTES} минут.
|
||
|
||
Если вы не запрашивали этот код, проигнорируйте это письмо.
|
||
|
||
С уважением,
|
||
Команда AutonetSellCar
|
||
"""
|
||
}
|
||
|
||
subject = subjects.get(language, subjects["en"])
|
||
body = bodies.get(language, bodies["en"])
|
||
|
||
# Check if SMTP is configured
|
||
if not settings.SMTP_USER or not settings.SMTP_PASSWORD:
|
||
# Development mode - just log the code
|
||
print(f"[DEV] Email verification code for {email}: {verification.code}")
|
||
return True, "Verification code sent (dev mode)"
|
||
|
||
# Send actual email
|
||
msg = MIMEMultipart()
|
||
msg['From'] = f"{settings.SMTP_FROM_NAME} <{settings.SMTP_FROM_EMAIL or settings.SMTP_USER}>"
|
||
msg['To'] = email
|
||
msg['Subject'] = subject
|
||
msg.attach(MIMEText(body, 'plain', 'utf-8'))
|
||
|
||
with smtplib.SMTP(settings.SMTP_HOST, settings.SMTP_PORT) as server:
|
||
server.starttls()
|
||
server.login(settings.SMTP_USER, settings.SMTP_PASSWORD)
|
||
server.send_message(msg)
|
||
|
||
return True, "Verification code sent to your email"
|
||
|
||
except Exception as e:
|
||
print(f"[ERROR] Failed to send email: {e}")
|
||
return False, f"Failed to send email: {str(e)}"
|
||
|
||
|
||
async def send_sms_verification(
|
||
db: Session,
|
||
phone: str,
|
||
user_id: Optional[int] = None,
|
||
language: str = "en"
|
||
) -> Tuple[bool, str]:
|
||
"""Send SMS verification code"""
|
||
# Normalize phone number
|
||
phone = phone.strip().replace(" ", "").replace("-", "")
|
||
if not phone.startswith("+"):
|
||
# Assume Mongolia if no country code
|
||
if phone.startswith("9") and len(phone) == 8:
|
||
phone = "+976" + phone
|
||
|
||
# Check rate limit (1 SMS per minute)
|
||
recent = db.query(VerificationCode).filter(
|
||
VerificationCode.phone == phone,
|
||
VerificationCode.code_type == "phone",
|
||
VerificationCode.created_at > datetime.utcnow() - timedelta(minutes=1)
|
||
).first()
|
||
|
||
if recent:
|
||
return False, "Please wait 1 minute before requesting another code."
|
||
|
||
# Create verification code
|
||
verification = create_verification_code(
|
||
db=db,
|
||
code_type="phone",
|
||
phone=phone,
|
||
user_id=user_id
|
||
)
|
||
|
||
# SMS messages by language
|
||
messages = {
|
||
"en": f"AutonetSellCar verification code: {verification.code}. Valid for {settings.VERIFICATION_CODE_EXPIRE_MINUTES} min.",
|
||
"ko": f"AutonetSellCar 인증 코드: {verification.code}. {settings.VERIFICATION_CODE_EXPIRE_MINUTES}분간 유효.",
|
||
"mn": f"AutonetSellCar баталгаажуулах код: {verification.code}. {settings.VERIFICATION_CODE_EXPIRE_MINUTES} мин хүчинтэй.",
|
||
"ru": f"Код подтверждения AutonetSellCar: {verification.code}. Действителен {settings.VERIFICATION_CODE_EXPIRE_MINUTES} мин."
|
||
}
|
||
|
||
message = messages.get(language, messages["en"])
|
||
|
||
try:
|
||
# Check if Twilio is configured
|
||
if not settings.TWILIO_ACCOUNT_SID or not settings.TWILIO_AUTH_TOKEN:
|
||
# Development mode - just log the code
|
||
print(f"[DEV] SMS verification code for {phone}: {verification.code}")
|
||
return True, "Verification code sent (dev mode)"
|
||
|
||
# Send actual SMS via Twilio
|
||
from twilio.rest import Client
|
||
|
||
client = Client(settings.TWILIO_ACCOUNT_SID, settings.TWILIO_AUTH_TOKEN)
|
||
client.messages.create(
|
||
body=message,
|
||
from_=settings.TWILIO_PHONE_NUMBER,
|
||
to=phone
|
||
)
|
||
|
||
return True, "Verification code sent to your phone"
|
||
|
||
except Exception as e:
|
||
print(f"[ERROR] Failed to send SMS: {e}")
|
||
return False, f"Failed to send SMS: {str(e)}"
|
||
|
||
|
||
def mark_email_verified(db: Session, user: User) -> None:
|
||
"""Mark user's email as verified"""
|
||
user.email_verified = True
|
||
user.email_verified_at = datetime.utcnow()
|
||
db.commit()
|
||
|
||
|
||
def mark_phone_verified(db: Session, user: User, phone: str) -> None:
|
||
"""Mark user's phone as verified and update phone number"""
|
||
user.phone = phone
|
||
user.phone_verified = True
|
||
user.phone_verified_at = datetime.utcnow()
|
||
db.commit()
|
||
|
||
|
||
def is_email_verified(user: User) -> bool:
|
||
"""Check if user's email is verified"""
|
||
return user.email_verified
|
||
|
||
|
||
def is_phone_verified(user: User) -> bool:
|
||
"""Check if user's phone is verified"""
|
||
return user.phone_verified
|