""" 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 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}" html_content = f"""

ACCÈS AU SIMULATEUR VORTEX FINANCE

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

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