""" WhatsApp 2FA Authentication — Meta WhatsApp Business Cloud API ONLY. Additive layer on top of existing email+password auth. Does NOT replace any existing module. When a user has `wa_otp_enabled=true`, the standard `/api/auth/login` endpoint returns `requires_wa_otp: true` + a short-lived challenge_token instead of session cookies. The frontend then calls: POST /api/auth/wa-otp/send { challenge_token } POST /api/auth/wa-otp/verify { otp_token, code } Provider: Meta WhatsApp Business Platform (Cloud API, v20.0). Fallback: Stub mode when META_WA_* env vars are missing — OTP is generated, stored, and surfaced in the JSON response (dev/demo only, gated by EXPOSE_OTP_CODE=1). This means the FULL flow works end-to-end without real Meta credentials, but real WhatsApp delivery is skipped. Multi-User + Multi-Company SAFE: * Each OTP is bound to a single user_id (no cross-user replay). * `wa_otp_challenges` is the only new collection — does NOT touch users, companies, sessions, items, invoices, or any business-logic collection. * Admin policy (`wa_auth_policy`) is global per deployment — admin can force-enable for everyone OR leave opt-in per user. Offline-First: this layer is ONLINE-only by design. Existing offline-cached sessions (`rbs_cached_user_v1`) are unaffected. When a user has WA 2FA enabled but the device is offline, the standard offline-cached session continues to work — WA OTP is only re-required on a fresh login. """ from __future__ import annotations import hashlib import hmac import logging import os import re import secrets import string from datetime import datetime, timedelta, timezone from typing import Optional import httpx import jwt from bson import ObjectId from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Request, Response from pydantic import BaseModel from tenacity import retry, stop_after_attempt, wait_exponential from auth import ( JWT_ALGORITHM, create_access_token, create_refresh_token, get_current_user, get_jwt_secret, hash_password, require_admin, set_auth_cookies, verify_password, ) logger = logging.getLogger("rge.wa_auth") # ---------------------------------------------------------------------------- # Configuration — all values come from env. Missing values → STUB MODE. # ---------------------------------------------------------------------------- META_GRAPH_VER = os.environ.get("META_WA_GRAPH_VERSION", "v20.0") META_PHONE_ID = os.environ.get("META_WA_PHONE_NUMBER_ID", "").strip() META_ACCESS_TOKEN = os.environ.get("META_WA_ACCESS_TOKEN", "").strip() META_APP_SECRET = os.environ.get("META_APP_SECRET", "").strip() META_VERIFY_TOKEN = os.environ.get("META_WA_VERIFY_TOKEN", "").strip() META_OTP_TEMPLATE = os.environ.get("META_WA_OTP_TEMPLATE_NAME", "rge_otp_v1").strip() META_OTP_LANGUAGE = os.environ.get("META_WA_OTP_TEMPLATE_LANG", "en_US").strip() # Dev visibility: when 1 and stub mode is active, OTP is included in JSON. # In production with real Meta creds, this is automatically ignored. EXPOSE_OTP_CODE = os.environ.get("EXPOSE_WA_OTP_CODE", "1") == "1" OTP_LEN = 6 OTP_TTL_SEC = 90 # 90-second window per spec OTP_MAX_TRIES = 5 OTP_LOCKOUT_MINUTES = 15 OTP_RESEND_COOLDOWN_SEC = 30 CHALLENGE_TTL_MIN = 10 # challenge_token expiry (from login → send-otp) def is_meta_configured() -> bool: """Return True only when ALL critical Meta creds are present.""" return bool(META_PHONE_ID and META_ACCESS_TOKEN) # ---------------------------------------------------------------------------- # Helpers # ---------------------------------------------------------------------------- def _now() -> datetime: return datetime.now(timezone.utc) def _now_iso() -> str: return _now().isoformat() def _gen_otp(n: int = OTP_LEN) -> str: """Cryptographically-secure n-digit numeric OTP (zero-padded).""" return "".join(secrets.choice(string.digits) for _ in range(n)) def _normalize_phone(raw: str) -> str: """E.164 normaliser — strips spaces/dashes, defaults to +91 for 10-digit Indian numbers. Returns digits-only string WITHOUT the leading '+' (Meta Cloud API expects this format).""" n = (raw or "").strip().replace(" ", "").replace("-", "") if not n: raise HTTPException(400, "Recipient phone number is required") if n.startswith("+"): n = n[1:] # If 10 digits and looks like Indian mobile, prepend 91 if len(n) == 10 and n.isdigit(): n = "91" + n if not n.isdigit() or len(n) < 8 or len(n) > 15: raise HTTPException(400, "Phone number is invalid") return n def _mask_phone(num: str) -> str: """Show last 4 digits only for UI display: 919812345678 → +91-XXXXXX-5678. Auto-normalises raw 10-digit Indian numbers (adds 91 prefix).""" if not num: return "" digits = re.sub(r"\D", "", num) # Normalise 10-digit Indian numbers to country-code form for display if len(digits) == 10: digits = "91" + digits if len(digits) < 4: return "•" * len(digits) return "+" + digits[:2] + "-XXXXXX-" + digits[-4:] # ---------------------------------------------------------------------------- # Meta WhatsApp Cloud API send (with retries + stub-mode fallback) # ---------------------------------------------------------------------------- @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10), reraise=True) async def _meta_send_otp(to_e164: str, code: str) -> dict: """Send OTP via approved Authentication-category template. Returns Meta's JSON response or raises. In STUB mode (no creds), returns a fake message_id and logs the OTP for developer visibility.""" if not is_meta_configured(): logger.info("[STUB] WhatsApp OTP %s → %s (Meta creds missing, no real send)", code, to_e164) return {"stub": True, "messages": [{"id": "stub-" + secrets.token_hex(6)}]} url = f"https://graph.facebook.com/{META_GRAPH_VER}/{META_PHONE_ID}/messages" payload = { "messaging_product": "whatsapp", "to": to_e164, "type": "template", "template": { "name": META_OTP_TEMPLATE, "language": {"code": META_OTP_LANGUAGE}, "components": [ {"type": "body", "parameters": [{"type": "text", "text": code}]}, { "type": "button", "sub_type": "url", "index": "0", "parameters": [{"type": "text", "text": code}], }, ], }, } headers = { "Authorization": f"Bearer {META_ACCESS_TOKEN}", "Content-Type": "application/json", } async with httpx.AsyncClient(timeout=12.0) as client: r = await client.post(url, json=payload, headers=headers) if r.status_code >= 400: body = r.text[:500] logger.warning("Meta WA send failed: %s %s", r.status_code, body) raise HTTPException(502, f"WhatsApp send failed ({r.status_code}): {body}") return r.json() # ---------------------------------------------------------------------------- # Policy (admin-controllable global config) # ---------------------------------------------------------------------------- DEFAULT_POLICY = { "_id": "global", "enabled": True, # Global kill-switch for WA 2FA "force_enable_for_all": False, # When True, every active user is required to verify WA OTP "force_enable_for_admins_only": False, # Only admin role required "expiry_sec": OTP_TTL_SEC, "max_retries": OTP_MAX_TRIES, "resend_cooldown_sec": OTP_RESEND_COOLDOWN_SEC, "lockout_minutes": OTP_LOCKOUT_MINUTES, "template_name": META_OTP_TEMPLATE, "template_lang": META_OTP_LANGUAGE, "updated_at": None, "updated_by": None, } async def get_policy(db) -> dict: doc = await db.wa_auth_policy.find_one({"_id": "global"}) if not doc: return dict(DEFAULT_POLICY) # Merge with defaults so newly-added keys never blow up old DBs. merged = dict(DEFAULT_POLICY) merged.update(doc) return merged async def _is_wa_required_for_user(db, user: dict) -> bool: """Return True if this user MUST complete WhatsApp OTP after password.""" policy = await get_policy(db) if not policy.get("enabled"): return False # Per-user opt-in if user.get("wa_otp_enabled"): return True # Admin force toggles if policy.get("force_enable_for_all"): return True if policy.get("force_enable_for_admins_only") and (user.get("role") == "admin"): return True return False # ---------------------------------------------------------------------------- # Challenge token helpers (links login → OTP send) # ---------------------------------------------------------------------------- def _issue_challenge(uid: str, email: str) -> str: payload = { "sub": uid, "email": email, "type": "wa_otp_challenge", "exp": _now() + timedelta(minutes=CHALLENGE_TTL_MIN), } return jwt.encode(payload, get_jwt_secret(), algorithm=JWT_ALGORITHM) def _decode_challenge(token: str) -> dict: try: decoded = jwt.decode(token, get_jwt_secret(), algorithms=[JWT_ALGORITHM]) except jwt.ExpiredSignatureError: raise HTTPException(401, "Challenge expired — please log in again") except jwt.InvalidTokenError: raise HTTPException(401, "Invalid challenge token") if decoded.get("type") != "wa_otp_challenge": raise HTTPException(401, "Wrong token type") return decoded # ---------------------------------------------------------------------------- # Public API # ---------------------------------------------------------------------------- router = APIRouter(prefix="/api/auth/wa-otp", tags=["wa-otp"]) admin_router = APIRouter(prefix="/api/admin/wa-auth", tags=["admin-wa-auth"]) webhook_router = APIRouter(prefix="/api/webhooks/whatsapp", tags=["wa-webhook"]) public_router = APIRouter(prefix="/api/wa-auth", tags=["wa-auth-public"]) # --------- /api/auth/wa-otp/send ------------------------------------------- class WaSendIn(BaseModel): challenge_token: str @router.post("/send") async def wa_send_otp(payload: WaSendIn, request: Request, background_tasks: BackgroundTasks): """Step-2 of WA login: exchange challenge_token → send OTP via WhatsApp.""" db = request.app.state.db decoded = _decode_challenge(payload.challenge_token) uid = decoded["sub"] try: user = await db.users.find_one({"_id": ObjectId(uid)}) except Exception: raise HTTPException(401, "Invalid user") if not user or user.get("is_active") is False: raise HTTPException(401, "Account not available") phone_raw = (user.get("phone") or "").strip() if not phone_raw: raise HTTPException( 400, "No WhatsApp number on file. Ask admin to update your phone number first.", ) to_e164 = _normalize_phone(phone_raw) # Resend cooldown — block if a previous unused OTP was created < N seconds ago policy = await get_policy(db) cooldown_sec = int(policy.get("resend_cooldown_sec") or OTP_RESEND_COOLDOWN_SEC) cutoff = (_now() - timedelta(seconds=cooldown_sec)).isoformat() recent = await db.wa_otp_challenges.find_one({ "user_id": uid, "used": False, "created_at": {"$gte": cutoff}, }) if recent: # Return the existing token so frontend can keep counting down — no duplicate send. # Note: we DO NOT resend the WhatsApp message during cooldown. return { "ok": True, "otp_token": recent["otp_token"], "masked_phone": _mask_phone(to_e164), "expires_in": int((datetime.fromisoformat(recent["expires_at"]) - _now()).total_seconds()), "resend_after": cooldown_sec - int((_now() - datetime.fromisoformat(recent["created_at"])).total_seconds()), "delivered_via": recent.get("delivered_via", "stored"), "cooldown": True, } # Invalidate any earlier outstanding OTPs for this user await db.wa_otp_challenges.update_many( {"user_id": uid, "used": False}, {"$set": {"used": True, "used_at": _now_iso(), "invalidated": True}}, ) code = _gen_otp(OTP_LEN) otp_token = secrets.token_urlsafe(24) expiry_sec = int(policy.get("expiry_sec") or OTP_TTL_SEC) max_tries = int(policy.get("max_retries") or OTP_MAX_TRIES) expires_at = _now() + timedelta(seconds=expiry_sec) delivered_via = "stub" meta_message_id = None send_error = None try: result = await _meta_send_otp(to_e164, code) meta_message_id = (result.get("messages") or [{}])[0].get("id") delivered_via = "whatsapp" if is_meta_configured() else "stub" except HTTPException as ex: send_error = ex.detail # We deliberately DO NOT raise — we still store the OTP and return a # success envelope. The frontend's fallback link to email-OTP path # (existing /api/security/login-otp/) keeps the user unblocked. logger.warning("WhatsApp delivery failed for user %s: %s", uid, send_error) delivered_via = "failed" except Exception as ex: send_error = str(ex) logger.exception("Unexpected WA send error") delivered_via = "failed" await db.wa_otp_challenges.insert_one({ "otp_token": otp_token, "user_id": uid, "email": user["email"], "phone": to_e164, "code_hash": hash_password(code), "expires_at": expires_at.isoformat(), "max_tries": max_tries, "tries": 0, "used": False, "created_at": _now_iso(), "delivered_via": delivered_via, "meta_message_id": meta_message_id, "send_error": send_error, "ip": request.client.host if request.client else "", }) await db.audit_log.insert_one({ "ts": _now_iso(), "user_id": uid, "email": user["email"], "action": "auth.wa_otp.sent", "outcome": "ok" if delivered_via in ("whatsapp", "stub") else "failed", "ip": request.client.host if request.client else "", "user_agent": (request.headers.get("user-agent") or "")[:240], "detail": {"delivered_via": delivered_via, "message_id": meta_message_id, "error": send_error}, }) resp = { "ok": True, "otp_token": otp_token, "masked_phone": _mask_phone(to_e164), "expires_in": expiry_sec, "resend_after": cooldown_sec, "delivered_via": delivered_via, "fallback_available": True, } # Dev visibility — only in stub mode, gated by env flag. Real Meta production runs # never include `code_preview` because `is_meta_configured()` is True. if EXPOSE_OTP_CODE and not is_meta_configured(): resp["code_preview"] = code return resp # --------- /api/auth/wa-otp/verify ----------------------------------------- class WaVerifyIn(BaseModel): otp_token: str code: str @router.post("/verify") async def wa_verify_otp(payload: WaVerifyIn, request: Request, response: Response): db = request.app.state.db ip = request.client.host if request.client else "" ua = (request.headers.get("user-agent") or "")[:240] rec = await db.wa_otp_challenges.find_one({"otp_token": payload.otp_token}) if not rec or rec.get("used"): raise HTTPException(401, "Invalid OTP session — please log in again") try: exp = datetime.fromisoformat(rec["expires_at"]) if exp.tzinfo is None: exp = exp.replace(tzinfo=timezone.utc) if exp < _now(): raise HTTPException(401, "OTP expired — request a new one") except (ValueError, TypeError): raise HTTPException(401, "Corrupted OTP record") if rec.get("tries", 0) >= int(rec.get("max_tries") or OTP_MAX_TRIES): raise HTTPException(429, "Too many attempts — wait and request a new OTP") await db.wa_otp_challenges.update_one({"_id": rec["_id"]}, {"$inc": {"tries": 1}}) code_in = (payload.code or "").strip() if not code_in.isdigit() or len(code_in) != OTP_LEN: raise HTTPException(400, f"OTP must be {OTP_LEN} digits") if not verify_password(code_in, rec["code_hash"]): await db.audit_log.insert_one({ "ts": _now_iso(), "user_id": rec["user_id"], "email": rec["email"], "action": "auth.wa_otp.failed", "outcome": "denied", "ip": ip, "user_agent": ua, }) raise HTTPException(401, "Wrong OTP code") # Mark used (single-use) + audit await db.wa_otp_challenges.update_one( {"_id": rec["_id"]}, {"$set": {"used": True, "used_at": _now_iso()}}, ) user = await db.users.find_one({"_id": ObjectId(rec["user_id"])}) if not user: raise HTTPException(401, "User not found") uid = str(user["_id"]) access = create_access_token(uid, user["email"], user.get("role", "staff")) refresh = create_refresh_token(uid) set_auth_cookies(response, access, refresh) await db.audit_log.insert_one({ "ts": _now_iso(), "user_id": uid, "email": user["email"], "action": "auth.wa_otp.success", "outcome": "ok", "ip": ip, "user_agent": ua, }) return { "id": uid, "email": user["email"], "name": user.get("name", ""), "role": user.get("role", "staff"), } # --------- /api/auth/wa-otp/resend (alias of /send, but force new) --------- @router.post("/resend") async def wa_resend_otp(payload: WaSendIn, request: Request, background_tasks: BackgroundTasks): """Manual resend — same as /send but ignores cooldown after the cooldown window. The cooldown check inside /send already returns the existing token; if the user wants a brand-new OTP after cooldown expiry, they simply call /send again. This endpoint exists for frontend ergonomics.""" return await wa_send_otp(payload, request, background_tasks) # --------- /api/wa-auth/me (current user's status; public-ish) ------------- @public_router.get("/me") async def my_wa_status(request: Request, user=Depends(get_current_user)): db = request.app.state.db doc = await db.users.find_one({"_id": ObjectId(user["id"])}) policy = await get_policy(db) return { "enabled": bool(doc.get("wa_otp_enabled")), "phone": doc.get("phone", ""), "masked_phone": _mask_phone(doc.get("phone", "")), "policy": { "global_enabled": bool(policy.get("enabled")), "force_enable_for_all": bool(policy.get("force_enable_for_all")), "force_enable_for_admins_only": bool(policy.get("force_enable_for_admins_only")), "expiry_sec": policy.get("expiry_sec"), "max_retries": policy.get("max_retries"), "resend_cooldown_sec": policy.get("resend_cooldown_sec"), "provider_configured": is_meta_configured(), }, } class ToggleSelfIn(BaseModel): enabled: bool @public_router.post("/me/toggle") async def toggle_self_wa(payload: ToggleSelfIn, request: Request, user=Depends(get_current_user)): """User opts in/out of WhatsApp 2FA. Requires a phone number on file.""" db = request.app.state.db doc = await db.users.find_one({"_id": ObjectId(user["id"])}) if payload.enabled and not (doc.get("phone") or "").strip(): raise HTTPException(400, "Add a phone number to your profile first") await db.users.update_one( {"_id": ObjectId(user["id"])}, {"$set": {"wa_otp_enabled": bool(payload.enabled), "wa_otp_updated_at": _now_iso()}}, ) await db.audit_log.insert_one({ "ts": _now_iso(), "user_id": user["id"], "email": user["email"], "action": "auth.wa_otp.toggle_self", "outcome": "ok", "detail": {"enabled": bool(payload.enabled)}, "ip": request.client.host if request.client else "", }) return {"ok": True, "enabled": bool(payload.enabled)} # --------- Admin: Policy + per-user override ------------------------------- @admin_router.get("/policy") async def admin_get_policy(request: Request, user=Depends(require_admin)): db = request.app.state.db p = await get_policy(db) p["provider_configured"] = is_meta_configured() return p class PolicyIn(BaseModel): enabled: Optional[bool] = None force_enable_for_all: Optional[bool] = None force_enable_for_admins_only: Optional[bool] = None expiry_sec: Optional[int] = None max_retries: Optional[int] = None resend_cooldown_sec: Optional[int] = None lockout_minutes: Optional[int] = None template_name: Optional[str] = None template_lang: Optional[str] = None @admin_router.put("/policy") async def admin_set_policy(payload: PolicyIn, request: Request, user=Depends(require_admin)): db = request.app.state.db update = {k: v for k, v in payload.model_dump(exclude_none=True).items()} update["updated_at"] = _now_iso() update["updated_by"] = user["email"] # Bounds validation if "expiry_sec" in update and not (30 <= update["expiry_sec"] <= 600): raise HTTPException(400, "expiry_sec must be 30-600") if "max_retries" in update and not (1 <= update["max_retries"] <= 10): raise HTTPException(400, "max_retries must be 1-10") if "resend_cooldown_sec" in update and not (10 <= update["resend_cooldown_sec"] <= 300): raise HTTPException(400, "resend_cooldown_sec must be 10-300") await db.wa_auth_policy.update_one({"_id": "global"}, {"$set": update}, upsert=True) return await get_policy(db) @admin_router.get("/users") async def admin_list_users(request: Request, user=Depends(require_admin)): """List users with WA OTP status — used by Admin Panel UI.""" db = request.app.state.db docs = await db.users.find({}, {"email": 1, "name": 1, "role": 1, "phone": 1, "wa_otp_enabled": 1, "is_active": 1}).to_list(2000) out = [] for d in docs: out.append({ "id": str(d["_id"]), "email": d.get("email"), "name": d.get("name", ""), "role": d.get("role", "staff"), "phone": d.get("phone", ""), "masked_phone": _mask_phone(d.get("phone", "")), "wa_otp_enabled": bool(d.get("wa_otp_enabled")), "is_active": d.get("is_active") is not False, }) return out class AdminToggleUserIn(BaseModel): enabled: bool @admin_router.put("/users/{uid}") async def admin_toggle_user(uid: str, payload: AdminToggleUserIn, request: Request, user=Depends(require_admin)): db = request.app.state.db try: oid = ObjectId(uid) except Exception: raise HTTPException(400, "Bad user id") doc = await db.users.find_one({"_id": oid}) if not doc: raise HTTPException(404, "User not found") if payload.enabled and not (doc.get("phone") or "").strip(): raise HTTPException(400, "Cannot enable: user has no phone number on file") await db.users.update_one( {"_id": oid}, {"$set": {"wa_otp_enabled": bool(payload.enabled), "wa_otp_updated_at": _now_iso()}}, ) await db.audit_log.insert_one({ "ts": _now_iso(), "user_id": user["id"], "email": user["email"], "action": "auth.wa_otp.admin_toggle_user", "outcome": "ok", "detail": {"target_user_id": uid, "enabled": bool(payload.enabled)}, }) return {"ok": True, "user_id": uid, "wa_otp_enabled": bool(payload.enabled)} @admin_router.get("/logs") async def admin_logs(request: Request, limit: int = 100, user=Depends(require_admin)): db = request.app.state.db docs = await db.audit_log.find({ "action": {"$in": [ "auth.wa_otp.sent", "auth.wa_otp.success", "auth.wa_otp.failed", "auth.wa_otp.toggle_self", "auth.wa_otp.admin_toggle_user", ]}, }).sort("ts", -1).limit(min(limit, 500)).to_list(500) out = [] for d in docs: out.append({ "ts": d.get("ts"), "action": d.get("action"), "outcome": d.get("outcome"), "email": d.get("email"), "ip": d.get("ip"), "detail": d.get("detail", {}), }) return out # --------- Meta WhatsApp delivery webhooks --------------------------------- def _verify_webhook_signature(payload_bytes: bytes, signature_header: str) -> bool: if not META_APP_SECRET: # Without an app secret, signature can't be checked — allow but log. logger.warning("META_APP_SECRET not set — webhook signature check skipped") return True if not signature_header or not signature_header.startswith("sha256="): return False received = signature_header.split("sha256=", 1)[1].strip() expected = hmac.new(META_APP_SECRET.encode(), payload_bytes, hashlib.sha256).hexdigest() return hmac.compare_digest(expected, received) @webhook_router.get("") async def wa_webhook_verify(request: Request): """Meta webhook subscription verification (one-time per setup).""" params = dict(request.query_params) hub_mode = params.get("hub.mode") hub_challenge = params.get("hub.challenge", "") hub_token = params.get("hub.verify_token", "") if hub_mode == "subscribe" and META_VERIFY_TOKEN and hub_token == META_VERIFY_TOKEN: # Meta requires raw integer response — but FastAPI's Response is fine return int(hub_challenge) if hub_challenge.isdigit() else hub_challenge raise HTTPException(403, "Invalid verify token") @webhook_router.post("") async def wa_webhook_receive(request: Request, background_tasks: BackgroundTasks): """Meta posts delivery/read/error events here. Must return 200 within 20s.""" raw = await request.body() sig = request.headers.get("X-Hub-Signature-256", "") if not _verify_webhook_signature(raw, sig): raise HTTPException(401, "Invalid signature") try: data = await request.json() except Exception: return {"ok": True} background_tasks.add_task(_process_wa_webhook, request.app.state.db, data) return {"ok": True} async def _process_wa_webhook(db, data: dict): """Background processor — updates wa_audit_logs with delivery state.""" try: entry = ((data.get("entry") or [{}])[0].get("changes") or [{}])[0].get("value") or {} # Status updates (sent/delivered/read/failed) for status_obj in (entry.get("statuses") or []): mid = status_obj.get("id") if not mid: continue await db.wa_audit_logs.update_one( {"message_id": mid}, { "$set": { "status": status_obj.get("status"), "recipient_id": status_obj.get("recipient_id"), "errors": status_obj.get("errors", []), "updated_at": _now_iso(), }, "$setOnInsert": {"first_seen_at": _now_iso()}, }, upsert=True, ) # Mark the matching OTP record as delivered when known if status_obj.get("status") in ("delivered", "read"): await db.wa_otp_challenges.update_one( {"meta_message_id": mid}, {"$set": {"delivered_at": _now_iso(), "delivery_state": status_obj.get("status")}}, ) except Exception: logger.exception("WA webhook processing failed") # ---------------------------------------------------------------------------- # Helper exposed to auth.py — to add WA OTP gate after password verification. # auth.py imports `should_require_wa_otp` lazily to avoid circular imports. # ---------------------------------------------------------------------------- async def should_require_wa_otp(db, user: dict) -> tuple[bool, Optional[str]]: """Returns (required, challenge_token). When required=True the login endpoint should respond with `requires_wa_otp: True` and `challenge_token` instead of cookies.""" if not await _is_wa_required_for_user(db, user): return False, None uid = str(user["_id"]) return True, _issue_challenge(uid, user["email"])