""" MODULE SIM_GATE — Barriere d'acces au simulateur public par lien email ========================================================================= Le simulateur public (client_simulator.py) etait jusqu'ici accessible a n'importe qui connaissant l'URL, sans aucune verification. Ce module ajoute une premiere barriere : pour y acceder, TOUT visiteur (sans exception, y compris l'administrateur) doit indiquer son email et cliquer sur un lien a usage unique recu par mail. Une fois le token valide, si l'email correspond a un client existant dans Clients_KYC, client_simulator.py saute la recherche manuelle d'identite et identifie directement ce client. A noter (limite assumee) : l'email auto-declare reste un indice, pas une preuve cryptographique d'identite — cette barriere decourage un acces trivial au lien public, elle ne remplace pas une authentification forte. """ import re import secrets as _secrets_lib import threading import time from datetime import datetime, timedelta import requests import streamlit as st from core.email_client import get_credential as _get_credential, DEFAULT_APP_BASE_URL REOUVERTURE_HEURES = 2 SIM_ACCESS_SHEET = "Simulateur_Acces_Tokens" SIM_ACCESS_HEADERS = ["Token", "Email", "Date_Envoi", "Date_Utilisation", "Statut"] # Anti-abus : la page est publique, chaque demande envoie un vrai email depuis le compte Brevo. LIMITE_PAR_EMAIL = 3 # liens par adresse et par heure LIMITE_GLOBALE = 40 # liens par heure, tous visiteurs confondus FENETRE_SECONDES = 3600 _RE_EMAIL = re.compile(r"[^@\s]{1,64}@[^@\s]{1,190}\.[^@\s]{2,}") @st.cache_resource def _registre_envois(): """Envois recents, partages par toutes les sessions du processus : [(horodatage, email)].""" return [], threading.Lock() def verifier_quota(email, maintenant=None, registre=None): """(autorise, motif). N'enregistre rien : appeler enregistrer_envoi() apres un envoi reussi.""" envois, verrou = registre or _registre_envois() maintenant = maintenant or time.time() cle = str(email or "").strip().lower() with verrou: envois[:] = [(t, e) for t, e in envois if maintenant - t < FENETRE_SECONDES] if sum(1 for _, e in envois if e == cle) >= LIMITE_PAR_EMAIL: return False, "Trop de liens demandés pour cette adresse. Réessayez dans une heure." if len(envois) >= LIMITE_GLOBALE: return False, "Service momentanément saturé. Réessayez dans une heure." return True, "" def enregistrer_envoi(email, maintenant=None, registre=None): envois, verrou = registre or _registre_envois() with verrou: envois.append((maintenant or time.time(), str(email or "").strip().lower())) def _ensure_sim_access_ws(sh): try: return sh.worksheet(SIM_ACCESS_SHEET) except Exception: ws = sh.add_worksheet(title=SIM_ACCESS_SHEET, rows=1000, cols=len(SIM_ACCESS_HEADERS)) ws.append_row(SIM_ACCESS_HEADERS) return ws def _email_client_connu(sh, sheet_name, email): """True/False selon que l'email figure dans Clients_KYC (colonne Email), None si la lecture echoue (429 compris) : l'appelant envoie alors le mail d'acces normal, sans jamais bloquer. Lecture en cache (1 appel groupe).""" try: from core import sheets_cache df = sheets_cache.get_cached_dfs_batch(sh, sheet_name, ["Clients_KYC"])["Clients_KYC"] if df.empty: return False if "Email" not in df.columns: return None cle = str(email or "").strip().lower() return bool((df["Email"].astype(str).str.strip().str.lower() == cle).any()) except Exception as exc: from core.errors import log_swallowed log_swallowed("Recherche de l'email dans Clients_KYC (simulateur) echouee", exc) return None def _creer_token_onboarding(sh, email, role="CLIENT"): """Cree un token d'onboarding identique a celui de client_simulator.envoyer_lien_onboarding (meme feuille, meme ligne), SANS envoyer d'email. Retourne le token, ou None en cas d'echec.""" try: from modules.client_simulator import _ensure_onboarding_ws token = _secrets_lib.token_urlsafe(24) now_str = datetime.now().strftime("%d-%m-%Y %H:%M:%S") _ensure_onboarding_ws(sh).append_row([token, email, now_str, "", "ENVOYE", role]) return token except Exception as exc: from core.errors import log_swallowed log_swallowed(f"Creation du token onboarding pour {email} (simulateur) echouee", exc) return None def envoyer_lien_simulateur(gspread_client, sheet_name, email_cible: str): """Envoie un lien d'acces et retourne ``(succes, detail)``.""" email_cible = str(email_cible or "").strip() if len(email_cible) > 254 or not _RE_EMAIL.fullmatch(email_cible): return False, "Adresse e-mail invalide." autorise, motif = verifier_quota(email_cible) if not autorise: return False, motif api_key = _get_credential("BREVO_API_KEY") sender_email = _get_credential("SENDER_EMAIL") base_url = _get_credential("APP_BASE_URL", DEFAULT_APP_BASE_URL) if not api_key: return False, "BREVO_API_KEY est absent des secrets." if not sender_email: return False, "SENDER_EMAIL est absent des secrets." token = _secrets_lib.token_urlsafe(24) now_str = datetime.now().strftime("%d-%m-%Y %H:%M:%S") try: sh = gspread_client.open(sheet_name) ws = _ensure_sim_access_ws(sh) ws.append_row([token, email_cible.strip(), now_str, "", "ENVOYE"]) except Exception as exc: return False, f"Impossible d'enregistrer le lien dans Google Sheets : {exc}" lien = f"{base_url.rstrip('/')}/?simaccess={token}" # Email inconnu de Clients_KYC : on joint directement le lien de creation # de profil (evite un 2e aller-retour email). Lecture en echec -> mail normal. # L'ecran affiche le meme message dans tous les cas (confidentialite). bloc_onboarding = "" if _email_client_connu(sh, sheet_name, email_cible) is False: token_onb = _creer_token_onboarding(sh, email_cible, role="CLIENT") if token_onb: lien_onb = f"{base_url.rstrip('/')}/?onboard=1&token={token_onb}" bloc_onboarding = f"""

Vous n'avez pas encore de profil : créez-le directement ici, puis revenez au simulateur.

Ce lien d'inscription est valable 48 heures. Préparez vos pièces : pièce d'identité, justificatif d'adresse, bulletin de salaire, RIB, relevés bancaires ou Wave/Orange Money.

""" html_content = f"""

ACCÈS AU SIMULATEUR VORTEX FINANCE

Veuillez accéder au simulateur de prêt en cliquant sur ce lien.

{bloc_onboarding}

Ce lien est personnel. En cas d'interruption, vous pouvez le rouvrir pendant 2 heures après sa première ouverture.

Si vous n'êtes pas à l'origine de cette demande, vous pouvez ignorer ce message.

""" url = "https://api.brevo.com/v3/smtp/email" headers = {"accept": "application/json", "api-key": api_key, "content-type": "application/json"} payload = { "sender": {"email": sender_email, "name": "Vortex Finance"}, "to": [{"email": email_cible}], "subject": "Votre lien d'accès au simulateur Vortex Finance", "htmlContent": html_content, } try: response = requests.post(url, json=payload, headers=headers, timeout=15) if response.status_code not in (200, 201): return False, f"Brevo a refusé l'envoi ({response.status_code}). Vérifiez l'adresse expéditrice et la clé API." enregistrer_envoi(email_cible) return True, "Lien envoyé." except Exception as e: from core.errors import log_swallowed log_swallowed(f"Envoi de l'email d'acces simulateur a {email_cible} echoue", e) return False, f"Connexion au service e-mail impossible : {e}" def consommer_token_simulateur(gspread_client, sheet_name, token: str): """Valide un token d'acces a usage unique. Retourne l'email associe (et marque le token comme utilise) s'il est valide et pas deja consomme, None sinon.""" if not token: return None try: sh = gspread_client.open(sheet_name) ws = _ensure_sim_access_ws(sh) records = ws.get_all_records() for i in range(len(records) - 1, -1, -1): rec = records[i] if str(rec.get("Token", "")).strip() != str(token).strip(): continue if str(rec.get("Statut", "")).strip() == "UTILISE": # Fenetre de reouverture : un client dont la page a ete # actualisee/rechargee (session perdue) peut rouvrir son lien # peu apres la 1re ouverture, sans devoir en redemander un. try: date_util = datetime.strptime(str(rec.get("Date_Utilisation", "")).strip(), "%d-%m-%Y %H:%M:%S") if datetime.now() - date_util <= timedelta(hours=REOUVERTURE_HEURES): return str(rec.get("Email", "")).strip() except ValueError: pass return None now_str = datetime.now().strftime("%d-%m-%Y %H:%M:%S") ws.update(f"D{i + 2}:E{i + 2}", [[now_str, "UTILISE"]]) return str(rec.get("Email", "")).strip() return None except Exception: return None