feat(auth): Implement Email OTP Verification

.env.example

@@ -49,4 +49,8 @@ FACE_ENCRYPTION_KEY=hkbribvfirirbvivbibvib
 CORS_ORIGINS=["http://localhost:3000", "http://localhost:5173", "http://127.0.0.1:3000", "http://127.0.0.1:5173"]
 
 # Firebase
-FIREBASE_CREDENTIALS_PATH=path/to/firebase-credentials.json
+FIREBASE_CREDENTIALS_PATH=path/to/firebase-credentials.json
+
+# Resend Email Configuration
+RESEND_API_KEY=
+EMAIL_FROM=

app/core/config.py

@@ -75,6 +75,10 @@ class Settings(BaseSettings):
     FACE_ENCRYPTION_KEY: str
     FIREBASE_CREDENTIALS_PATH: str
 
+    # Resend Email Configuration
+    RESEND_API_KEY: str = ""
+    EMAIL_FROM: str = "onboarding@resend.dev"
+
     model_config = SettingsConfigDict(
         env_file=".env",
         extra="ignore",

app/infra/email.py

@@ -0,0 +1,61 @@
+import json
+import urllib.request
+import urllib.error
+import asyncio
+from app.core.config import settings
+from app.core.logger import logger
+
+class EmailSender:
+    @staticmethod
+    async def send_otp_email(to_email: str, otp: str) -> bool:
+        if not settings.RESEND_API_KEY:
+            logger.warning("RESEND_API_KEY is not set. Skipping email sending.")
+            # During development without an API key, just log the OTP
+            logger.info("MOCK EMAIL to %s: Your OTP is %s", to_email, otp)
+            return True
+
+        url = "https://api.resend.com/emails"
+        headers = {
+            "Authorization": f"Bearer {settings.RESEND_API_KEY}",
+            "Content-Type": "application/json",
+            "User-Agent": "multAI-Backend/1.0"
+        }
+
+        html_content = f"""
+        <div style="font-family: Arial, sans-serif; max-width: 600px; margin: 0 auto;">
+            <h2>Bienvenue sur multAI !</h2>
+            <p>Voici votre code de vérification :</p>
+            <h1 style="background: #f4f4f4; padding: 10px; letter-spacing: 5px; text-align: center;">{otp}</h1>
+            <p>Ce code est valide pendant 10 minutes.</p>
+        </div>
+        """
+
+        data = {
+            "from": settings.EMAIL_FROM,
+            "to": [to_email],
+            "subject": "Votre code de vérification multAI",
+            "html": html_content
+        }
+
+        req = urllib.request.Request(
+            url,
+            data=json.dumps(data).encode("utf-8"),
+            headers=headers,
+            method="POST"
+        )
+
+        def _send() -> bool:
+            try:
+                with urllib.request.urlopen(req, timeout=10) as response:
+                    res_body = response.read()
+                    logger.info("Email sent via Resend. Response: %s", res_body)
+                    return True
+            except urllib.error.HTTPError as e:
+                err_body = e.read()
+                logger.error("Failed to send email via Resend: %s - %s", e.code, err_body)
+                return False
+            except Exception as e:
+                logger.error("Error sending email: %s", str(e))
+                return False
+
+        return await asyncio.to_thread(_send)

app/router/mobile/auth.py

@@ -11,11 +11,12 @@
 from app.schema.request.mobile.auth import (
     MobileLoginRequest,
     MobileRegisterRequest,
+    RegisterVerifyRequest,
     RefreshTokenRequest,
     UpdateDeviceTokenRequest,
     InactivateDeviceRequest,
 )
-from app.schema.response.mobile.auth import MeResponse, DeviceSchema, MobileAuthResponse, SessionSchema, UserSchema
+from app.schema.response.mobile.auth import MeResponse, DeviceSchema, MobileAuthResponse, SessionSchema, UserSchema, RegisterPendingResponse
 
 router = APIRouter(prefix="/auth")
 
@@ -33,18 +34,29 @@ def _get_client_ip(request: Request) -> str | None:
     return request.client.host if request.client else None
 
 
-@router.post("/register", response_model=MobileAuthResponse)
+@router.post("/register", response_model=RegisterPendingResponse)
 async def mobile_register(
     req: MobileRegisterRequest,
     request: Request,
     container: Container = Depends(get_container),
-) -> MobileAuthResponse:
+) -> RegisterPendingResponse:
     client_ip = _get_client_ip(request)
     result = await container.auth_service.mobile_register(container.redis, req, client_ip=client_ip)
+    return result
+
+
+@router.post("/register/verify", response_model=MobileAuthResponse)
+async def mobile_register_verify(
+    req: RegisterVerifyRequest,
+    request: Request,
+    container: Container = Depends(get_container),
+) -> MobileAuthResponse:
+    client_ip = _get_client_ip(request)
+    result = await container.auth_service.verify_mobile_register(container.redis, req, client_ip=client_ip)
     await container.audit_service.create_record(
         event_type=AuditEventType.USER_SIGNUP,
         user_id=result.user_id,
-        metadata={"endpoint": "register"},
+        metadata={"endpoint": "register_verify"},
     )
     return result
 

app/schema/request/mobile/auth.py

@@ -63,6 +63,10 @@ class MobileLoginRequest(MobileAuthBaseRequest):
     pass
 
 
+class RegisterVerifyRequest(MobileAuthBaseRequest):
+    otp: str = Field(..., min_length=6, max_length=6, description="The 6-digit OTP code sent via email")
+
+
 
 
 

app/schema/response/mobile/auth.py

@@ -25,6 +25,10 @@ class MeResponse(BaseModel):
     sessions: Optional[SessionSchema]
 
 
+class RegisterPendingResponse(BaseModel):
+    message: str
+    status: str
+    email: str
 
 class MobileAuthResponse(BaseModel):
     access_token: str

app/service/users.py

@@ -22,8 +22,12 @@
     MobileAuthBaseRequest,
     MobileLoginRequest,
     MobileRegisterRequest,
+    RegisterVerifyRequest,
 )
-from app.schema.response.mobile.auth import MobileAuthResponse
+from app.schema.response.mobile.auth import MobileAuthResponse, RegisterPendingResponse
+from app.infra.nats import NatsClient
+import secrets
+import json
 from db.generated import user as user_queries
 from db.generated import devices as device_queries
 from db.generated import session as session_queries
@@ -131,7 +135,7 @@ async def mobile_register(
         redis: RedisClient,
         req: MobileRegisterRequest,
         client_ip: Optional[str] = None,
-    ) -> MobileAuthResponse:
+    ) -> RegisterPendingResponse:
         logger.info("mobile_register attempt")
         max_attempts = settings.RATE_LIMIT_LOGIN_MAX_ATTEMPTS
         window = settings.RATE_LIMIT_LOGIN_WINDOW_SECONDS
@@ -154,16 +158,61 @@ async def mobile_register(
         if existing_user is not None:
             logger.warning("register attempt: email_already_in_use")
             raise AppException.conflict("Email already in use; please login instead")
+
         hashed = hash_password(req.password)
-        logger.info("register attempt: creating_new_user")
+        otp = "".join(secrets.choice("0123456789") for _ in range(6))
+
+        pending_key = f"pending_user:{req.email}"
+        pending_data = {
+            "hashed_password": hashed,
+        }
+
+        # Save in Redis for 10 minutes (600 seconds)
+        await redis.set(pending_key, json.dumps(pending_data), expire=600)
+        await redis.set(f"otp:{req.email}", otp, expire=600)
+
+        # Send to NATS
+        await NatsClient.publish("email.send_otp", json.dumps({"email": req.email, "otp": otp}).encode("utf-8"))
+
+        logger.info("register success, OTP sent to %s", req.email)
+        return RegisterPendingResponse(
+            message="OTP sent to email",
+            status="pending_verification",
+            email=req.email
+        )
+
+    async def verify_mobile_register(
+        self,
+        redis: RedisClient,
+        req: RegisterVerifyRequest,
+        client_ip: Optional[str] = None,
+    ) -> MobileAuthResponse:
+        otp_key = f"otp:{req.email}"
+        stored_otp = await redis.get(otp_key)
+
+        if not stored_otp or stored_otp != req.otp:
+            raise AppException.unauthorized("Invalid or expired OTP")
+
+        pending_key = f"pending_user:{req.email}"
+        raw_data = await redis.get(pending_key)
+        if not raw_data:
+            raise AppException.unauthorized("Registration session expired")
+
+        data = json.loads(raw_data)
+
         try:
-            user = await self.user_querier.create_user(email=req.email, hashed_password=hashed)
+            user = await self.user_querier.create_user(email=req.email, hashed_password=data["hashed_password"])
             if not user:
                 raise AppException.internal_error("Failed to create user")
         except SQLAlchemyError as exc:
             logger.error("Failed to create user: %s", exc)
             raise DBException.handle(exc)
-        logger.info("register success user_id=%s", user.id)
+
+        # Clean up redis
+        await redis.delete(otp_key)
+        await redis.delete(pending_key)
+
+        logger.info("register verify success user_id=%s", user.id)
         return await self._create_mobile_session(
             redis=redis,
             user=user,

app/worker/email_worker/main.py

@@ -0,0 +1,62 @@
+import asyncio
+import json
+
+from app.core.config import settings
+from app.core.logger import logger
+from app.infra.nats import NatsClient
+from app.infra.email import EmailSender
+
+
+async def handle_message(raw_payload: bytes | str) -> None:
+    try:
+        if isinstance(raw_payload, bytes):
+            raw_payload = raw_payload.decode()
+
+        payload = json.loads(raw_payload)
+        email = payload.get("email")
+        otp = payload.get("otp")
+
+        if not email or not otp:
+            logger.error("Invalid email.send_otp payload: %s", raw_payload)
+            return
+
+        success = await EmailSender.send_otp_email(to_email=email, otp=otp)
+        if success:
+            logger.info("Successfully sent OTP email to %s", email)
+        else:
+            logger.error("Failed to send OTP email to %s", email)
+
+    except Exception:
+        logger.exception("Unexpected error in email worker")
+
+
+async def run_worker() -> None:
+    logger.info("Email worker started")
+
+    async def wrapped_handler(msg: bytes | str) -> None:
+        await handle_message(msg)
+
+    # Subscribe to the email.send_otp subject
+    await NatsClient.subscribe("email.send_otp", wrapped_handler)
+
+    # Keep the worker running
+    await asyncio.Event().wait()
+
+
+async def main() -> None:
+    await NatsClient.connect(
+        host=settings.NATS_HOST,
+        port=settings.NATS_PORT,
+        user=settings.NATS_USER,
+        password=settings.NATS_PASSWORD,
+    )
+
+    try:
+        await run_worker()
+    finally:
+        await NatsClient.close()
+        logger.info("Email Worker shutdown")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

docker-compose.staging.local.yml

@@ -14,5 +14,13 @@ services:
       dockerfile: Dockerfile
     pull_policy: never
 
+  email-worker:
+    build:
+      context: .
+      dockerfile: Dockerfile
+    pull_policy: never
+    volumes:
+      - ./:/app
+
 volumes:
   insightface_cache:

docker-compose.staging.yml

@@ -96,6 +96,19 @@ services:
     networks:
       - multi_network
 
+  email-worker:
+    image: ghcr.io/microclub-usthb/multai-back:latest
+    container_name: multi_email_worker
+    restart: unless-stopped
+    env_file:
+      - .env.staging
+    depends_on:
+      - nats
+      - redis
+    command: ["uv", "run", "python", "-m", "app.worker.email_worker.main"]
+    networks:
+      - multi_network
+
 volumes:
   postgres_data:
   minio_data:

pyproject.toml

@@ -31,6 +31,7 @@ dependencies = [
     "opencv-python>=4.13.0.92",
     "filetype>=1.2.0",
     "pillow-heif>=1.3.0",
+    "sqlalchemy>=2.0.47",
 ]
 
 [tool.ruff]