refactor: promo and email bruteforce defence

bot/services/promo_code_service.py

@@ -1,10 +1,12 @@
 import logging
+from html import escape as html_escape
 from datetime import datetime
 from sqlalchemy.ext.asyncio import AsyncSession
 from typing import Optional, Tuple, Dict
 from aiogram import Bot
 
 from config.settings import Settings
+from db.dal import security_dal
 
 from db.dal import promo_code_dal, user_dal
 from db.models import PromoCode, User
@@ -24,6 +26,9 @@ class PromoCodeService:
         self.bot = bot
         self.i18n = i18n
 
+    def _throttle_identifier(self, user_id: int) -> str:
+        return f"user:{int(user_id)}"
+
     async def apply_promo_code(
         self,
         session: AsyncSession,
@@ -32,19 +37,45 @@ class PromoCodeService:
         user_lang: str,
     ) -> Tuple[bool, datetime | str]:
         _ = lambda k, **kw: self.i18n.gettext(user_lang, k, **kw)
-        code_input_upper = code_input.strip().upper()
+        code_input_upper = (code_input or "").strip().upper()[:100]
+        code_display = html_escape(code_input_upper[:100], quote=False)
+        throttle_identifier = self._throttle_identifier(user_id)
+
+        throttle = await security_dal.check_throttle(
+            session,
+            scope=security_dal.PROMO_CODE_APPLY_SCOPE,
+            identifier=throttle_identifier,
+        )
+        if throttle.locked:
+            return False, _(
+                "promo_code_too_many_attempts",
+                seconds=throttle.retry_after or max(1, int(self.settings.BRUTE_FORCE_LOCK_SECONDS)),
+            )
 
         promo_data = await promo_code_dal.get_active_promo_code_by_code_str(
             session, code_input_upper)
 
         if not promo_data:
-            return False, _("promo_code_not_found", code=code_input_upper)
+            throttle_result = await security_dal.record_throttle_failure(
+                session,
+                scope=security_dal.PROMO_CODE_APPLY_SCOPE,
+                identifier=throttle_identifier,
+                max_failures=self.settings.BRUTE_FORCE_MAX_FAILURES,
+                window_seconds=self.settings.BRUTE_FORCE_WINDOW_SECONDS,
+                lock_seconds=self.settings.BRUTE_FORCE_LOCK_SECONDS,
+            )
+            if throttle_result.locked:
+                return False, _(
+                    "promo_code_too_many_attempts",
+                    seconds=throttle_result.retry_after or max(1, int(self.settings.BRUTE_FORCE_LOCK_SECONDS)),
+                )
+            return False, _("promo_code_not_found", code=code_display)
 
         existing_activation = await promo_code_dal.get_user_activation_for_promo(
             session, promo_data.promo_code_id, user_id)
         if existing_activation:
             return False, _("promo_code_already_used_by_user",
-                            code=code_input_upper)
+                            code=code_display)
 
         bonus_days = promo_data.bonus_days
 
@@ -61,6 +92,11 @@ class PromoCodeService:
                 session, promo_data.promo_code_id)
 
             if activation_recorded and promo_incremented:
+                await security_dal.clear_throttle_state(
+                    session,
+                    scope=security_dal.PROMO_CODE_APPLY_SCOPE,
+                    identifier=throttle_identifier,
+                )
                 # Send notification about promo activation
                 try:
                     notification_service = NotificationService(self.bot, self.settings, self.i18n)