"""Authentication module: JWT, bcrypt, brute force protection, auth routes.""" import os import bcrypt import jwt from datetime import datetime, timezone, timedelta from fastapi import APIRouter, HTTPException, Request, Response, Depends from pydantic import BaseModel, EmailStr from typing import Optional JWT_ALGORITHM = "HS256" ACCESS_TOKEN_MIN = 60 * 12 # 12 hours for ERP UX REFRESH_TOKEN_DAYS = 30 LOCKOUT_THRESHOLD = 5 LOCKOUT_MINUTES = 15 # Admin-controllable: when False, login lockout is disabled entirely (failed # attempts are still audited, but the account is never blocked). Default OFF # per user requirement — uninterrupted billing workflow. Toggle from Admin # Security page → "Auto Account Lock" switch. LOCKOUT_ENABLED = os.environ.get("AUTH_LOCKOUT_ENABLED", "false").lower() == "true" def get_jwt_secret() -> str: return os.environ["JWT_SECRET"] def hash_password(password: str) -> str: return bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8") def verify_password(plain: str, hashed: str) -> bool: try: return bcrypt.checkpw(plain.encode("utf-8"), hashed.encode("utf-8")) except Exception: return False def create_access_token(user_id: str, email: str, role: str) -> str: payload = { "sub": user_id, "email": email, "role": role, "exp": datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_MIN), "type": "access", } return jwt.encode(payload, get_jwt_secret(), algorithm=JWT_ALGORITHM) def create_refresh_token(user_id: str) -> str: payload = { "sub": user_id, "exp": datetime.now(timezone.utc) + timedelta(days=REFRESH_TOKEN_DAYS), "type": "refresh", } return jwt.encode(payload, get_jwt_secret(), algorithm=JWT_ALGORITHM) def set_auth_cookies(response: Response, access: str, refresh: str): response.set_cookie("access_token", access, httponly=True, secure=True, samesite="none", max_age=ACCESS_TOKEN_MIN * 60, path="/") response.set_cookie("refresh_token", refresh, httponly=True, secure=True, samesite="none", max_age=REFRESH_TOKEN_DAYS * 86400, path="/") def clear_auth_cookies(response: Response): # Cookies must be deleted with the SAME attributes used to set them # (path, samesite, secure) — otherwise browsers ignore the delete. for key in ("access_token", "refresh_token"): response.set_cookie(key, "", httponly=True, secure=True, samesite="none", max_age=0, path="/", expires=0) # ---- pydantic schemas ---- class LoginInput(BaseModel): email: EmailStr password: str class RegisterInput(BaseModel): email: EmailStr password: str name: str role: str = "staff" # 'admin' or 'staff' # Extended optional fields (collected on the public signup page) first_name: Optional[str] = "" last_name: Optional[str] = "" company_name: Optional[str] = "" phone: Optional[str] = "" gstin: Optional[str] = "" address: Optional[str] = "" state: Optional[str] = "" business_type: Optional[str] = "" class UserOut(BaseModel): id: str email: str name: str role: str # ---- DB-backed helpers ---- async def _find_user(db, email: str): return await db.users.find_one({"email": email.lower()}) async def _user_to_out(doc) -> dict: return { "id": str(doc["_id"]), "email": doc["email"], "name": doc.get("name", ""), "role": doc.get("role", "staff"), } async def get_current_user(request: Request) -> dict: db = request.app.state.db token = request.cookies.get("access_token") if not token: auth_header = request.headers.get("Authorization", "") if auth_header.startswith("Bearer "): token = auth_header[7:] if not token: raise HTTPException(status_code=401, detail="Not authenticated") try: payload = jwt.decode(token, get_jwt_secret(), algorithms=[JWT_ALGORITHM]) if payload.get("type") != "access": raise HTTPException(status_code=401, detail="Invalid token type") from bson import ObjectId try: user = await db.users.find_one({"_id": ObjectId(payload["sub"])}) except Exception: raise HTTPException(status_code=401, detail="Invalid token subject") if not user: raise HTTPException(status_code=401, detail="User not found") if user.get("is_active") is False: raise HTTPException(status_code=403, detail="Account is inactive") # Session revocation: if admin called revoke-sessions, all tokens issued # before that wall-clock instant are rejected. invalid_before = user.get("tokens_invalid_before") if invalid_before: try: iat = datetime.fromtimestamp(payload.get("iat", 0), tz=timezone.utc).isoformat() if iat < invalid_before: raise HTTPException(status_code=401, detail="Session revoked. Please log in again.") except HTTPException: raise except Exception: pass return await _user_to_out(user) except jwt.ExpiredSignatureError: raise HTTPException(status_code=401, detail="Token expired") except jwt.InvalidTokenError: raise HTTPException(status_code=401, detail="Invalid token") async def require_admin(user: dict = Depends(get_current_user)) -> dict: if user.get("role") != "admin": raise HTTPException(status_code=403, detail="Admin access required") return user # ---- brute force ---- def _ensure_aware(dt): """Mongo BSON datetimes round-trip as naive UTC. Treat as UTC if no tz set.""" if dt is None: return None try: return dt.replace(tzinfo=timezone.utc) if dt.tzinfo is None else dt except AttributeError: return None # In-process cache of the auth-policy doc to avoid hitting Mongo on every login. # Refreshed on every PUT through `/api/auth/security-policy`. _POLICY_CACHE: dict = {"lockout_enabled": LOCKOUT_ENABLED, "idle_timeout_min": 0, "loaded": False} def _is_lockout_enabled(_db) -> bool: """Whether to enforce auto-lockout on this server. The DB-level policy (set by an admin via Admin Security page) wins over the environment-variable default. When the cache hasn't been loaded yet (cold start), fall back to the env default. """ return bool(_POLICY_CACHE.get("lockout_enabled", LOCKOUT_ENABLED)) async def _load_policy_cache(db) -> None: """Refresh the in-process policy cache from Mongo.""" doc = await db.auth_policy.find_one({"_id": "global"}) or {} _POLICY_CACHE["lockout_enabled"] = bool(doc.get("lockout_enabled", LOCKOUT_ENABLED)) _POLICY_CACHE["idle_timeout_min"] = int(doc.get("idle_timeout_min", 0)) _POLICY_CACHE["loaded"] = True async def _check_lockout(db, ident: str): # Skip the lockout check entirely when disabled. Admins can still see # failed-attempt counts in `db.login_attempts` for audit purposes — we # just never block the user from logging in. if not _is_lockout_enabled(db): return rec = await db.login_attempts.find_one({"_id": ident}) if not rec: return if rec.get("count", 0) >= LOCKOUT_THRESHOLD: locked_until = _ensure_aware(rec.get("locked_until")) now = datetime.now(timezone.utc) if locked_until and locked_until > now: mins = int((locked_until - now).total_seconds() // 60) + 1 raise HTTPException(status_code=429, detail=f"Too many failed attempts. Try again in {mins} minutes.") # Lockout expired — auto-reset so users don't stay locked out forever. if locked_until and locked_until <= now: await db.login_attempts.delete_one({"_id": ident}) async def _record_failure(db, ident: str): now = datetime.now(timezone.utc) rec = await db.login_attempts.find_one({"_id": ident}) count = (rec.get("count", 0) if rec else 0) + 1 update = {"count": count, "last_attempt": now} # Only set the locked_until timestamp when lockout is enabled — otherwise # we still track the count for audit visibility but never enforce a block. if _is_lockout_enabled(db) and count >= LOCKOUT_THRESHOLD: update["locked_until"] = now + timedelta(minutes=LOCKOUT_MINUTES) await db.login_attempts.update_one({"_id": ident}, {"$set": update}, upsert=True) async def _clear_attempts(db, ident: str): await db.login_attempts.delete_one({"_id": ident}) # ---- router ---- def get_auth_router(): router = APIRouter(prefix="/api/auth", tags=["auth"]) @router.post("/register") async def register(payload: RegisterInput, request: Request, response: Response): """Public self-signup. Creates a user + a default company (multi-tenant). First-ever signup automatically becomes admin so the very first account is never locked out of its own data. """ db = request.app.state.db ip = request.client.host if request.client else "unknown" ua = (request.headers.get("user-agent") or "")[:240] email = payload.email.lower().strip() # 1. Validate inputs (Pydantic already handles email format) if not payload.password or len(payload.password) < 8: raise HTTPException(status_code=422, detail="Password must be at least 8 characters") display_name = (payload.name or "").strip() or f"{(payload.first_name or '').strip()} {(payload.last_name or '').strip()}".strip() if not display_name: raise HTTPException(status_code=422, detail="Name is required") # 2. Reject duplicates if await _find_user(db, email): raise HTTPException(status_code=409, detail="An account with this email already exists. Try logging in instead.") # 3. Optional GST sanity (15 chars, alphanumeric) — format-only, not real-time lookup gstin = (payload.gstin or "").strip().upper() if gstin and (len(gstin) != 15 or not gstin.isalnum()): raise HTTPException(status_code=422, detail="GSTIN must be exactly 15 alphanumeric characters") # 4. First-user-becomes-admin policy (multi-tenant safe) user_count = await db.users.count_documents({}) assigned_role = "admin" if user_count == 0 else payload.role or "staff" now_iso = datetime.now(timezone.utc).isoformat() user_doc = { "email": email, "password_hash": hash_password(payload.password), "name": display_name, "first_name": (payload.first_name or "").strip(), "last_name": (payload.last_name or "").strip(), "company_name": (payload.company_name or "").strip(), "phone": (payload.phone or "").strip(), "gstin": gstin, "address": (payload.address or "").strip(), "state": (payload.state or "").strip(), "business_type": (payload.business_type or "").strip(), "role": assigned_role, "is_active": True, "twofa_enabled": False, "created_at": now_iso, "updated_at": now_iso, "signup_ip": ip, "signup_user_agent": ua, "plan": "free", "license_status": "trial", } res = await db.users.insert_one(user_doc) uid = str(res.inserted_id) # 5. Auto-create a default company so the user lands in a usable workspace if payload.company_name and payload.company_name.strip(): try: await db.companies.insert_one({ "name": payload.company_name.strip(), "owner_id": uid, "gstin": gstin, "address": (payload.address or "").strip(), "state": (payload.state or "").strip(), "phone": (payload.phone or "").strip(), "email": email, "industry": (payload.business_type or "").strip() or "General", "created_at": now_iso, }) except Exception as ex: print(f"[auth.register] default company creation failed: {ex}") # 6. Audit log + auto-login (set cookies) await db.audit_log.insert_one({ "ts": now_iso, "user_id": uid, "email": email, "action": "auth.register", "outcome": "ok", "ip": ip, "user_agent": ua, "detail": {"role": assigned_role, "company": payload.company_name}, }) access = create_access_token(uid, email, assigned_role) refresh = create_refresh_token(uid) set_auth_cookies(response, access, refresh) return { "id": uid, "email": email, "name": display_name, "role": assigned_role, "company_name": payload.company_name or "", "is_first_user": user_count == 0, } @router.post("/login") async def login(payload: LoginInput, request: Request, response: Response): db = request.app.state.db ip = request.client.host if request.client else "unknown" ua = (request.headers.get("user-agent") or "")[:240] email = payload.email.lower() ident = f"{ip}:{email}" await _check_lockout(db, ident) user = await _find_user(db, email) if not user or not verify_password(payload.password, user["password_hash"]): await _record_failure(db, ident) # Audit failed attempt await db.audit_log.insert_one({ "ts": datetime.now(timezone.utc).isoformat(), "user_id": str(user["_id"]) if user else "", "email": email, "action": "auth.login.failed", "outcome": "denied", "ip": ip, "user_agent": ua, "detail": {"reason": "invalid_credentials"}, }) raise HTTPException(status_code=401, detail="Invalid email or password") if user.get("is_active") is False: await db.audit_log.insert_one({ "ts": datetime.now(timezone.utc).isoformat(), "user_id": str(user["_id"]), "email": email, "action": "auth.login.failed", "outcome": "denied", "ip": ip, "user_agent": ua, "detail": {"reason": "inactive"}, }) raise HTTPException(status_code=403, detail="Account is inactive. Contact admin.") await _clear_attempts(db, ident) uid = str(user["_id"]) # WhatsApp 2FA gate — when user has wa_otp_enabled OR admin policy # force-enables it. This is checked BEFORE the TOTP gate so the two # do not conflict. If both happen to be enabled, WA OTP takes # priority (it's the user-facing primary 2FA channel for ERPs in India). try: from whatsapp_auth import should_require_wa_otp required, challenge = await should_require_wa_otp(db, user) except Exception as ex: print(f"[auth] wa-otp gate skipped: {ex}") required, challenge = False, None if required and challenge: await db.audit_log.insert_one({ "ts": datetime.now(timezone.utc).isoformat(), "user_id": uid, "email": email, "action": "auth.wa_otp.challenge_issued", "outcome": "pending", "ip": ip, "user_agent": ua, }) # Mask phone for UX from whatsapp_auth import _mask_phone # type: ignore return { "requires_wa_otp": True, "challenge_token": challenge, "email": user["email"], "masked_phone": _mask_phone(user.get("phone", "")), } # 2FA gate — if enabled, return a short-lived challenge token instead. if user.get("twofa_enabled"): challenge = jwt.encode({ "sub": uid, "email": user["email"], "type": "2fa_challenge", "exp": datetime.now(timezone.utc) + timedelta(minutes=5), }, get_jwt_secret(), algorithm=JWT_ALGORITHM) await db.audit_log.insert_one({ "ts": datetime.now(timezone.utc).isoformat(), "user_id": uid, "email": email, "action": "auth.2fa.challenge_issued", "outcome": "pending", "ip": ip, "user_agent": ua, }) return {"requires_2fa": True, "challenge_token": challenge, "email": user["email"]} access = create_access_token(uid, user["email"], user.get("role", "staff")) refresh = create_refresh_token(uid) set_auth_cookies(response, access, refresh) # Audit + activity log await db.audit_log.insert_one({ "ts": datetime.now(timezone.utc).isoformat(), "user_id": uid, "email": email, "action": "auth.login.success", "outcome": "ok", "ip": ip, "user_agent": ua, }) await db.activity_logs.insert_one({ "user_id": uid, "user_email": user["email"], "action": "login", "entity": "auth", "timestamp": datetime.now(timezone.utc).isoformat(), }) # Best-effort suspicious-login detection (new IP / new browser). # NOTE: `security_engine` imports `verify_password`/`get_password_hash` from this # module, so we use a function-scope import here to avoid the import-time cycle. # This is the intentional Python pattern to break module-level cycles — do NOT # hoist this to the top of the file. try: from security_engine import detect_suspicious await detect_suspicious(db, {"id": uid, "email": user["email"]}, ip, ua) except Exception as ex: print(f"[auth] suspicious-login detection skipped: {ex}") out = await _user_to_out(user) if user.get("force_password_update"): out["force_password_update"] = True return out @router.post("/2fa/verify") async def login_2fa_verify(payload: dict, request: Request, response: Response): """Second step after the first login returns requires_2fa=true.""" import pyotp db = request.app.state.db ip = request.client.host if request.client else "unknown" ua = (request.headers.get("user-agent") or "")[:240] token = (payload or {}).get("challenge_token", "") code = (payload or {}).get("code", "").strip() try: decoded = jwt.decode(token, get_jwt_secret(), algorithms=[JWT_ALGORITHM]) except Exception: raise HTTPException(401, "Invalid or expired challenge token") if decoded.get("type") != "2fa_challenge": raise HTTPException(401, "Wrong token type") from bson import ObjectId user = await db.users.find_one({"_id": ObjectId(decoded["sub"])}) if not user or not user.get("twofa_enabled"): raise HTTPException(400, "2FA not enabled for this user") secret = user.get("twofa_secret", "") verified = bool(secret and pyotp.TOTP(secret).verify(code, valid_window=1)) # Recovery code fallback if not verified and code: for h in user.get("twofa_recovery_hashes", []) or []: if verify_password(code, h): verified = True # Burn the recovery code await db.users.update_one( {"_id": user["_id"]}, {"$pull": {"twofa_recovery_hashes": h}}, ) break if not verified: await db.audit_log.insert_one({ "ts": datetime.now(timezone.utc).isoformat(), "user_id": str(user["_id"]), "email": user["email"], "action": "auth.2fa.failed", "outcome": "denied", "ip": ip, "user_agent": ua, }) raise HTTPException(401, "Invalid 2FA code") 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": datetime.now(timezone.utc).isoformat(), "user_id": uid, "email": user["email"], "action": "auth.login.success", "outcome": "ok", "ip": ip, "user_agent": ua, "detail": {"twofa": True}, }) return await _user_to_out(user) @router.post("/logout") async def logout(response: Response): clear_auth_cookies(response) return {"ok": True} @router.get("/me") async def me(user=Depends(get_current_user)): return user # ----- Self-service password change (logged-in user with old password) ----- class ChangePwIn(BaseModel): old_password: str new_password: str @router.post("/change-password") async def change_password(payload: ChangePwIn, request: Request, response: Response, user=Depends(get_current_user)): """Authenticated user changes their own password. Why: This is different from /security/reset-password (forgot-password token flow) and /admin/users/{uid}/reset-password (admin-resets-someone- else). Here a logged-in user proves they know their old password and sets a new one. After success, the access cookie is rotated and all other refresh tokens are invalidated so the user is forced to re-login on other devices. """ db = request.app.state.db from bson import ObjectId from security_engine import score_password def _now_iso(): return datetime.now(timezone.utc).isoformat() old_pw = (payload.old_password or "").strip() new_pw = (payload.new_password or "").strip() if not old_pw or not new_pw: raise HTTPException(400, "Both old and new passwords are required") if old_pw == new_pw: raise HTTPException(400, "New password must be different from the old one") if len(new_pw) < 8: raise HTTPException(400, "New password must be at least 8 characters") # Validate strength using the same scorer that public flows use. score = score_password(new_pw) if score["score"] < 2: tips = "; ".join(score.get("suggestions") or ["use a stronger password"]) raise HTTPException(400, f"Password too weak — {tips}") # Look up the live user document — get_current_user only returns a slim shape. try: doc = await db.users.find_one({"_id": ObjectId(user["id"])}) except Exception: raise HTTPException(404, "User not found") if not doc: raise HTTPException(404, "User not found") if not verify_password(old_pw, doc.get("password_hash", "")): # Audit a failed attempt — useful signal for security review try: await db.audit_log.insert_one({ "ts": _now_iso(), "user_id": str(doc["_id"]), "email": doc["email"], "action": "auth.change_password.failed", "outcome": "wrong_old_password", "ip": request.client.host if request.client else "", "user_agent": (request.headers.get("user-agent") or "")[:240], }) except Exception: pass raise HTTPException(401, "Current password is incorrect") # Apply the new password + bookkeeping. try: await db.users.update_one( {"_id": doc["_id"]}, {"$set": { "password_hash": hash_password(new_pw), "password_changed_at": _now_iso(), "force_password_update": False, "is_temp_password": False, }}, ) except Exception as e: raise HTTPException(500, f"Failed to save new password: {e}") # Rotate the access cookie so the current tab keeps working immediately. new_access = create_access_token(str(doc["_id"]), doc["email"], doc.get("role", "staff")) response.set_cookie( "access_token", new_access, httponly=True, secure=True, samesite="none", max_age=ACCESS_TOKEN_MIN * 60, path="/", ) # Audit the successful change. try: await db.audit_log.insert_one({ "ts": _now_iso(), "user_id": str(doc["_id"]), "email": doc["email"], "action": "auth.change_password", "outcome": "ok", "ip": request.client.host if request.client else "", "user_agent": (request.headers.get("user-agent") or "")[:240], }) except Exception: pass return {"ok": True, "message": "Password updated successfully. Use the new password on other devices."} @router.post("/refresh") async def refresh(request: Request, response: Response): db = request.app.state.db token = request.cookies.get("refresh_token") if not token: raise HTTPException(status_code=401, detail="No refresh token") try: payload = jwt.decode(token, get_jwt_secret(), algorithms=[JWT_ALGORITHM]) if payload.get("type") != "refresh": raise HTTPException(status_code=401, detail="Invalid token type") from bson import ObjectId user = await db.users.find_one({"_id": ObjectId(payload["sub"])}) if not user: raise HTTPException(status_code=401, detail="User not found") access = create_access_token(str(user["_id"]), user["email"], user.get("role", "staff")) response.set_cookie("access_token", access, httponly=True, secure=True, samesite="none", max_age=ACCESS_TOKEN_MIN * 60, path="/") return {"ok": True} except jwt.ExpiredSignatureError: raise HTTPException(status_code=401, detail="Refresh token expired") except jwt.InvalidTokenError: raise HTTPException(status_code=401, detail="Invalid refresh token") # ----- Security Policy (admin-only) ----- @router.get("/security-policy") async def get_security_policy(request: Request, user=Depends(get_current_user)): """Read the current auth security policy. Returns defaults if not yet set.""" db = request.app.state.db if not _POLICY_CACHE.get("loaded"): await _load_policy_cache(db) return { "lockout_enabled": bool(_POLICY_CACHE.get("lockout_enabled", LOCKOUT_ENABLED)), "idle_timeout_min": int(_POLICY_CACHE.get("idle_timeout_min", 0)), "lockout_threshold": LOCKOUT_THRESHOLD, "lockout_minutes": LOCKOUT_MINUTES, } @router.put("/security-policy") async def update_security_policy(request: Request, user=Depends(get_current_user)): """Update auth security policy. Admin-only.""" if (user or {}).get("role") != "admin": raise HTTPException(status_code=403, detail="Admin access required") db = request.app.state.db body = await request.json() update = {} if "lockout_enabled" in body: update["lockout_enabled"] = bool(body["lockout_enabled"]) if "idle_timeout_min" in body: # 0 = never auto-logout; positive ints in minutes try: v = int(body["idle_timeout_min"]) update["idle_timeout_min"] = max(0, v) except (TypeError, ValueError): raise HTTPException(status_code=422, detail="idle_timeout_min must be a non-negative integer") if not update: raise HTTPException(status_code=422, detail="No policy fields supplied") update["updated_at"] = datetime.now(timezone.utc).isoformat() update["updated_by"] = user.get("email") or user.get("id") await db.auth_policy.update_one({"_id": "global"}, {"$set": update}, upsert=True) # Audit log try: await db.audit_log.insert_one({ "ts": update["updated_at"], "user_id": user.get("id"), "email": user.get("email"), "action": "auth.policy.update", "outcome": "ok", "details": update, }) except Exception as ex: print(f"[auth] policy audit log failed: {ex}") await _load_policy_cache(db) # If admin disabled lockout, also clear any in-flight blocks so users # who were stuck become immediately accessible again. if update.get("lockout_enabled") is False: await db.login_attempts.delete_many({}) return {"ok": True, "policy": _POLICY_CACHE} return router # ---- admin seeding ---- async def _ensure_admin(db, email: str, password: str, *, allow_legacy_migrate: bool = False): """Idempotently ensure an admin user exists at `email` with the given password. `allow_legacy_migrate=True` lets us migrate the legacy hard-coded admin (admin@rmregal.com or whatever single admin existed at greenfield) onto the new email. Only the PRIMARY admin uses this — secondary admins are always inserted fresh. """ email = (email or "").strip().lower() if not email or not password: return existing = await db.users.find_one({"email": email}) if existing: # Sync password to current env value to keep ops in lock-step. # Also clear force_password_update — env password IS the source of truth. updates = {"is_active": True} if not verify_password(password, existing["password_hash"]): updates["password_hash"] = hash_password(password) updates["role"] = "admin" if existing.get("force_password_update"): updates["force_password_update"] = False if existing.get("is_temp_password"): updates["is_temp_password"] = False await db.users.update_one({"_id": existing["_id"]}, {"$set": updates}) return if allow_legacy_migrate: # No user at the desired email — try to migrate a legacy admin in place. legacy = await db.users.find_one({"role": "admin"}) if legacy: await db.users.update_one( {"_id": legacy["_id"]}, {"$set": { "email": email, "password_hash": hash_password(password), "role": "admin", "is_active": True, "migrated_at": datetime.now(timezone.utc).isoformat(), }}, ) return # Greenfield install or secondary admin — fresh insert. await db.users.insert_one({ "email": email, "password_hash": hash_password(password), "name": "Administrator", "role": "admin", "is_active": True, "created_at": datetime.now(timezone.utc).isoformat(), }) async def seed_admin(db): """Idempotently ensure the env-configured admin account(s) exist. Reads: • ADMIN_EMAIL / ADMIN_PASSWORD — primary admin (required). Will migrate a legacy hard-coded admin in place if this email doesn't exist yet. • ADMIN2_EMAIL / ADMIN2_PASSWORD — optional secondary admin. No-op if either is blank. Both admins get role="admin" and is_active=True. Passwords are bcrypt-hashed. """ await _ensure_admin( db, os.environ.get("ADMIN_EMAIL", ""), os.environ.get("ADMIN_PASSWORD", ""), allow_legacy_migrate=True, ) await _ensure_admin( db, os.environ.get("ADMIN2_EMAIL", ""), os.environ.get("ADMIN2_PASSWORD", ""), allow_legacy_migrate=False, ) async def cleanup_demo_users(db): """One-shot purge of legacy demo accounts (staff@rmregal.com, admin@rmregal.com). Only runs if a real admin (from env) exists separately.""" admin_email = os.environ.get("ADMIN_EMAIL", "").strip().lower() if not admin_email: return real = await db.users.find_one({"email": admin_email}) if not real: return for demo_email in ("staff@rmregal.com", "admin@rmregal.com"): if demo_email == admin_email: continue await db.users.delete_one({"email": demo_email})