Spaces:
Running
Running
Claude Opus 5.5
fix(securite): XSS sur donnees publiques, limite d'envoi des liens simulateur
c4a01d3 | """ | |
| 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,}") | |
| 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""" | |
| <html><body style="font-family: 'Courier New', monospace; background-color: #0a0e12; color: #c9d1d9; padding: 20px;"> | |
| <div style="border: 1px solid #58a6ff; padding: 20px; background-color: rgba(10, 14, 18, 0.8);"> | |
| <h2 style="color: #58a6ff;">ACCÈS AU SIMULATEUR VORTEX FINANCE</h2> | |
| <p>Veuillez accéder au simulateur de prêt en <a href="{lien}" style="color: #58a6ff; text-decoration: underline; font-weight: bold;">cliquant sur ce lien</a>.</p> | |
| <p style="font-size: 0.8rem; color: #7a8a9a; margin-top: 20px;">Ce lien est personnel. En cas d'interruption, vous pouvez le rouvrir pendant 2 heures après sa première ouverture.</p> | |
| <p style="font-size: 0.8rem; color: #7a8a9a;">Si vous n'êtes pas à l'origine de cette demande, vous pouvez ignorer ce message.</p> | |
| </div> | |
| </body></html> | |
| """ | |
| 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 | |