from flask_limiter import Limiter from flask_limiter.util import get_remote_address from flask_migrate import Migrate from flask_sqlalchemy import SQLAlchemy from flask_wtf.csrf import CSRFProtect def _rate_limit_breach_response(request_limit): import time from flask import jsonify, make_response, render_template_string, request, session, url_for from utils.decorators import is_admin_authenticated from services.security_abuse import record_rate_limit_event message = "Has hecho demasiadas solicitudes. Espera un momento e intenta de nuevo." accept = (request.headers.get("Accept") or "").lower() wants_json = request.headers.get("X-Requested-With") == "XMLHttpRequest" or "application/json" in accept retry_after_seconds = None retry_after_text = "Intenta de nuevo en aproximadamente 1 minuto." reset_at = getattr(request_limit, "reset_at", None) if reset_at: try: retry_after_seconds = max(int(reset_at) - int(time.time()), 1) minutes = max((retry_after_seconds + 59) // 60, 1) retry_after_text = f"Intenta de nuevo en aproximadamente {minutes} minuto{'s' if minutes != 1 else ''}." except (TypeError, ValueError): retry_after_seconds = None try: record_rate_limit_event( route_path=request.path or "unknown", endpoint=request.endpoint or "unknown", limit_label=str(request_limit.limit), ) except Exception: pass if wants_json: response = jsonify( { "ok": False, "error": message, "limit": str(request_limit.limit), "retry_after_text": retry_after_text, "retry_after_seconds": retry_after_seconds, } ) response.status_code = 429 response.headers["X-RateLimit-Handled"] = "1" response.headers["Cache-Control"] = "no-store" if retry_after_seconds: response.headers["Retry-After"] = str(retry_after_seconds) return response if request.path.startswith("/admin"): fallback = url_for("admin.dashboard") if is_admin_authenticated() else url_for("admin.login") elif session.get("user_id"): fallback = url_for("user.dashboard") elif request.path.startswith("/register"): fallback = url_for("auth.register") else: fallback = url_for("auth.login") html = render_template_string( """ Demasiadas solicitudes

Demasiadas solicitudes

{{ message }}

{{ retry_after_text }}

Si sigues viendo este aviso, vuelve al login y entra otra vez.

Volver
""", message=message, fallback=fallback, retry_after_text=retry_after_text, ) response = make_response(html, 429) if retry_after_seconds: response.headers["Retry-After"] = str(retry_after_seconds) response.headers["X-RateLimit-Handled"] = "1" response.headers["Cache-Control"] = "no-store" return response db = SQLAlchemy() csrf = CSRFProtect() migrate = Migrate() limiter = Limiter(key_func=get_remote_address, default_limits=[], on_breach=_rate_limit_breach_response)