""" 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