diff --git a/.env.example b/.env.example
index a953570..9ede870 100644
--- a/.env.example
+++ b/.env.example
@@ -57,6 +57,9 @@ SMTP_USE_SSL=False # U
EMAIL_CODE_TTL_SECONDS=600 # Email verification code lifetime
EMAIL_CODE_RESEND_SECONDS=60 # Minimum delay between code sends
EMAIL_CODE_MAX_ATTEMPTS=5 # Max attempts per code
+BRUTE_FORCE_MAX_FAILURES=5 # Max failed code attempts in the throttle window
+BRUTE_FORCE_WINDOW_SECONDS=900 # Rolling window used to count failures
+BRUTE_FORCE_LOCK_SECONDS=1800 # Temporary lockout duration after too many failures
# Payment Method Toggles
YOOKASSA_ENABLED=True # Turn on YOOKASSA
diff --git a/README.md b/README.md
index ed21019..73f478e 100644
--- a/README.md
+++ b/README.md
@@ -83,6 +83,11 @@
| `SMTP_USERNAME` / `SMTP_PASSWORD` | Логин и SMTP key/password из Brevo. Если не заданы вместе с `SMTP_FROM_EMAIL`, вход по email скрывается. | `user@smtp-brevo.com` |
| `SMTP_FROM_EMAIL` / `SMTP_FROM_NAME` | Подтвержденный отправитель и отображаемое имя отправителя для писем с кодом. | `no-reply@example.com` |
| `EMAIL_CODE_TTL_SECONDS` | Срок действия кода подтверждения email. | `600` |
+| `EMAIL_CODE_RESEND_SECONDS` | Минимальная пауза между отправками кода на один email. | `60` |
+| `EMAIL_CODE_MAX_ATTEMPTS` | Максимум попыток на один конкретный код. | `5` |
+| `BRUTE_FORCE_MAX_FAILURES` | Максимум неудачных попыток в окне защиты от перебора. | `5` |
+| `BRUTE_FORCE_WINDOW_SECONDS` | Длительность окна, в котором считаются неудачные попытки. | `900` |
+| `BRUTE_FORCE_LOCK_SECONDS` | Время временной блокировки после превышения лимита. | `1800` |
| `MY_DEVICES_SECTION_ENABLED` | Включить раздел «Мои устройства» в меню подписки (`true`/`false`). | `false` |
| `REQUIRED_CHANNEL_ID` | (Опционально) ID канала, на который пользователь должен подписаться перед использованием. Оставьте пустым, если проверка не нужна. | `-1001234567890` |
| `REQUIRED_CHANNEL_LINK` | (Опционально) Публичная ссылка или invite на канал для кнопки «Проверить подписку». | `https://t.me/your_channel` |
diff --git a/bot/app/web/subscription_webapp.py b/bot/app/web/subscription_webapp.py
index 604d120..33e78ba 100644
--- a/bot/app/web/subscription_webapp.py
+++ b/bot/app/web/subscription_webapp.py
@@ -426,8 +426,17 @@ async def email_auth_verify_route(request: web.Request) -> web.Response:
target_user_id=None,
)
if not verify_result.ok:
- await session.rollback()
- return _json_error(400, verify_result.error or "invalid_code", "Invalid code")
+ await session.commit()
+ status = 429 if verify_result.error == "rate_limited" else 400
+ return web.json_response(
+ {
+ "ok": False,
+ "error": verify_result.error or "invalid_code",
+ "retry_after": verify_result.retry_after,
+ "message": "Invalid code",
+ },
+ status=status,
+ )
db_user = await user_dal.get_user_by_email(session, email)
created_user = False
@@ -526,8 +535,17 @@ async def account_email_verify_route(request: web.Request) -> web.Response:
target_user_id=user_id,
)
if not verify_result.ok:
- await session.rollback()
- return _json_error(400, verify_result.error or "invalid_code", "Invalid code")
+ await session.commit()
+ status = 429 if verify_result.error == "rate_limited" else 400
+ return web.json_response(
+ {
+ "ok": False,
+ "error": verify_result.error or "invalid_code",
+ "retry_after": verify_result.retry_after,
+ "message": "Invalid code",
+ },
+ status=status,
+ )
current_user = await user_dal.get_user_by_id(session, user_id)
if not current_user or current_user.is_banned:
@@ -645,7 +663,7 @@ async def apply_promo_route(request: web.Request) -> web.Response:
lang,
)
if not success:
- await session.rollback()
+ await session.commit()
return _json_error(400, "promo_apply_failed", str(result))
await session.commit()
end_date = result if isinstance(result, datetime) else None
diff --git a/bot/handlers/user/promo_user.py b/bot/handlers/user/promo_user.py
index 40f6b6a..62e9045 100644
--- a/bot/handlers/user/promo_user.py
+++ b/bot/handlers/user/promo_user.py
@@ -4,7 +4,6 @@ from aiogram import Router, F, types, Bot
from aiogram.fsm.context import FSMContext
from typing import Optional
from sqlalchemy.ext.asyncio import AsyncSession
-from aiogram.utils.markdown import hcode
from config.settings import Settings
from bot.states.user_states import UserPromoStates
@@ -123,46 +122,41 @@ async def process_promo_code_input(message: types.Message, state: FSMContext,
except Exception as e:
logging.error(f"Failed to send suspicious promo notification: {e}")
- response_to_user_text = _("promo_code_not_found",
- code=hcode(code_input.upper()))
- reply_markup = get_back_to_main_menu_markup(current_lang, i18n)
+ success, result = await promo_code_service.apply_promo_code(
+ session, user.id, code_input, current_lang)
+ if success:
+ await session.commit()
+ logging.info(
+ f"Promo code '{code_input}' successfully applied for user {user.id}."
+ )
+
+ new_end_date = result if isinstance(result, datetime) else None
+ active = await subscription_service.get_active_subscription_details(session, user.id)
+ config_link_display = active.get("config_link") if active else None
+ connect_button_url = active.get("connect_button_url") if active else None
+ config_link_text = config_link_display or _("config_link_not_available")
+
+ response_to_user_text = _(
+ "promo_code_applied_success_full",
+ end_date=(new_end_date.strftime("%d.%m.%Y %H:%M:%S") if new_end_date else "N/A"),
+ config_link=config_link_text,
+ )
+ reply_markup = get_connect_and_main_keyboard(
+ current_lang,
+ i18n,
+ settings,
+ config_link_display,
+ connect_button_url=connect_button_url,
+ )
else:
-
- success, result = await promo_code_service.apply_promo_code(
- session, user.id, code_input, current_lang)
- if success:
- await session.commit()
- logging.info(
- f"Promo code '{code_input}' successfully applied for user {user.id}."
- )
-
- new_end_date = result if isinstance(result, datetime) else None
- active = await subscription_service.get_active_subscription_details(session, user.id)
- config_link_display = active.get("config_link") if active else None
- connect_button_url = active.get("connect_button_url") if active else None
- config_link_text = config_link_display or _("config_link_not_available")
-
- response_to_user_text = _(
- "promo_code_applied_success_full",
- end_date=(new_end_date.strftime("%d.%m.%Y %H:%M:%S") if new_end_date else "N/A"),
- config_link=config_link_text,
- )
- reply_markup = get_connect_and_main_keyboard(
- current_lang,
- i18n,
- settings,
- config_link_display,
- connect_button_url=connect_button_url,
- )
- else:
- await session.rollback()
- logging.info(
- f"Promo code '{code_input}' application failed for user {user.id}. Reason: {result}"
- )
- response_to_user_text = result
- reply_markup = get_back_to_main_menu_markup(
- current_lang, i18n
- )
+ await session.commit()
+ logging.info(
+ f"Promo code '{code_input}' application failed for user {user.id}. Reason: {result}"
+ )
+ response_to_user_text = result
+ reply_markup = get_back_to_main_menu_markup(
+ current_lang, i18n
+ )
await message.answer(
response_to_user_text,
diff --git a/bot/handlers/user/start.py b/bot/handlers/user/start.py
index 61c4386..0a72ca5 100644
--- a/bot/handlers/user/start.py
+++ b/bot/handlers/user/start.py
@@ -688,7 +688,7 @@ async def start_command_handler(message: types.Message,
# Don't show main menu if promo was successfully applied
return
else:
- await session.rollback()
+ await session.commit()
logging.warning(f"Failed to auto-apply promo code '{promo_code_to_apply}' for user {user_id}: {result}")
await message.answer(str(result), parse_mode="HTML")
# Continue to show main menu if promo failed
diff --git a/bot/services/email_auth_service.py b/bot/services/email_auth_service.py
index 9bb9201..af32432 100644
--- a/bot/services/email_auth_service.py
+++ b/bot/services/email_auth_service.py
@@ -16,6 +16,7 @@ from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from config.settings import Settings
+from db.dal import security_dal
from db.models import EmailVerificationCode
logger = logging.getLogger(__name__)
@@ -41,6 +42,7 @@ class EmailCodeRequestResult:
class EmailCodeVerifyResult:
ok: bool
error: Optional[str] = None
+ retry_after: Optional[int] = None
def normalize_email(value: str) -> str:
@@ -52,6 +54,11 @@ def is_valid_email(value: str) -> bool:
return bool(email and len(email) <= 254 and EMAIL_RE.match(email))
+def _email_throttle_identifier(email: str, purpose: str, target_user_id: Optional[int]) -> str:
+ target_part = "none" if target_user_id is None else str(target_user_id)
+ return f"{purpose}:{target_part}:{email}"
+
+
class EmailAuthService:
def __init__(self, settings: Settings):
self.settings = settings
@@ -106,6 +113,19 @@ class EmailAuthService:
return EmailCodeRequestResult(ok=False, error="invalid_email")
now = datetime.now(timezone.utc)
+ throttle = await security_dal.check_throttle(
+ session,
+ scope=security_dal.EMAIL_CODE_VERIFY_SCOPE,
+ identifier=_email_throttle_identifier(normalized_email, purpose, target_user_id),
+ now=now,
+ )
+ if throttle.locked:
+ return EmailCodeRequestResult(
+ ok=False,
+ error="rate_limited",
+ retry_after=throttle.retry_after,
+ )
+
latest_code = await self._get_latest_code(
session,
email=normalized_email,
@@ -154,9 +174,28 @@ class EmailAuthService:
) -> EmailCodeVerifyResult:
normalized_email = normalize_email(email)
normalized_code = re.sub(r"\D", "", code or "")
- if not is_valid_email(normalized_email) or len(normalized_code) != 6:
+ if not is_valid_email(normalized_email):
return EmailCodeVerifyResult(ok=False, error="invalid_code")
+ now = datetime.now(timezone.utc)
+ throttle_identifier = _email_throttle_identifier(
+ normalized_email,
+ purpose,
+ target_user_id,
+ )
+ throttle = await security_dal.check_throttle(
+ session,
+ scope=security_dal.EMAIL_CODE_VERIFY_SCOPE,
+ identifier=throttle_identifier,
+ now=now,
+ )
+ if throttle.locked:
+ return EmailCodeVerifyResult(
+ ok=False,
+ error="rate_limited",
+ retry_after=throttle.retry_after,
+ )
+
latest_code = await self._get_latest_code(
session,
email=normalized_email,
@@ -166,7 +205,6 @@ class EmailAuthService:
if not latest_code or latest_code.consumed_at is not None:
return EmailCodeVerifyResult(ok=False, error="invalid_code")
- now = datetime.now(timezone.utc)
expires_at = latest_code.expires_at
if expires_at.tzinfo is None:
expires_at = expires_at.replace(tzinfo=timezone.utc)
@@ -177,13 +215,55 @@ class EmailAuthService:
if int(latest_code.attempts or 0) >= max_attempts:
return EmailCodeVerifyResult(ok=False, error="too_many_attempts")
+ if len(normalized_code) != 6:
+ latest_code.attempts = int(latest_code.attempts or 0) + 1
+ throttle_result = await security_dal.record_throttle_failure(
+ session,
+ scope=security_dal.EMAIL_CODE_VERIFY_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,
+ now=now,
+ )
+ await session.flush()
+ if throttle_result.locked:
+ return EmailCodeVerifyResult(
+ ok=False,
+ error="rate_limited",
+ retry_after=throttle_result.retry_after,
+ )
+ if int(latest_code.attempts or 0) >= max_attempts:
+ return EmailCodeVerifyResult(ok=False, error="too_many_attempts")
+ return EmailCodeVerifyResult(ok=False, error="invalid_code")
+
expected_hash = self._hash_code(normalized_email, purpose, normalized_code)
if not hmac.compare_digest(expected_hash, latest_code.code_hash):
latest_code.attempts = int(latest_code.attempts or 0) + 1
+ throttle_result = await security_dal.record_throttle_failure(
+ session,
+ scope=security_dal.EMAIL_CODE_VERIFY_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,
+ now=now,
+ )
await session.flush()
+ if throttle_result.locked:
+ return EmailCodeVerifyResult(
+ ok=False,
+ error="rate_limited",
+ retry_after=throttle_result.retry_after,
+ )
return EmailCodeVerifyResult(ok=False, error="invalid_code")
latest_code.consumed_at = now
+ await security_dal.clear_throttle_state(
+ session,
+ scope=security_dal.EMAIL_CODE_VERIFY_SCOPE,
+ identifier=throttle_identifier,
+ )
await session.flush()
return EmailCodeVerifyResult(ok=True)
diff --git a/bot/services/promo_code_service.py b/bot/services/promo_code_service.py
index 77ec565..5421ae7 100644
--- a/bot/services/promo_code_service.py
+++ b/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)
diff --git a/config/settings.py b/config/settings.py
index 9440e87..fe3b590 100644
--- a/config/settings.py
+++ b/config/settings.py
@@ -228,6 +228,18 @@ class Settings(BaseSettings):
EMAIL_CODE_TTL_SECONDS: int = Field(default=10 * 60)
EMAIL_CODE_RESEND_SECONDS: int = Field(default=60)
EMAIL_CODE_MAX_ATTEMPTS: int = Field(default=5)
+ BRUTE_FORCE_MAX_FAILURES: int = Field(
+ default=5,
+ description="Maximum failed code attempts allowed within the throttle window before a temporary lockout is applied.",
+ )
+ BRUTE_FORCE_WINDOW_SECONDS: int = Field(
+ default=15 * 60,
+ description="Rolling window used to count failed email and promo code attempts.",
+ )
+ BRUTE_FORCE_LOCK_SECONDS: int = Field(
+ default=30 * 60,
+ description="Temporary lockout duration applied after too many failed code attempts.",
+ )
LOGS_PAGE_SIZE: int = Field(default=10)
diff --git a/db/dal/__init__.py b/db/dal/__init__.py
index 8df65b5..d85402a 100644
--- a/db/dal/__init__.py
+++ b/db/dal/__init__.py
@@ -6,6 +6,7 @@ from . import panel_sync_dal
from . import message_log_dal
from . import user_billing_dal
from . import ad_dal
+from . import security_dal
__all__ = (
"user_dal",
@@ -16,6 +17,7 @@ __all__ = (
"message_log_dal",
"user_billing_dal",
"ad_dal",
+ "security_dal",
)
diff --git a/db/dal/security_dal.py b/db/dal/security_dal.py
new file mode 100644
index 0000000..1662422
--- /dev/null
+++ b/db/dal/security_dal.py
@@ -0,0 +1,160 @@
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+from typing import Optional
+
+from sqlalchemy import case, delete, or_, select
+from sqlalchemy.dialects.postgresql import insert as pg_insert
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from ..models import SecurityThrottle
+
+EMAIL_CODE_VERIFY_SCOPE = "email_code_verify"
+PROMO_CODE_APPLY_SCOPE = "promo_code_apply"
+
+
+@dataclass(frozen=True)
+class ThrottleDecision:
+ locked: bool
+ retry_after: Optional[int] = None
+
+
+def _utc_now(value: Optional[datetime] = None) -> datetime:
+ if value is None:
+ value = datetime.now(timezone.utc)
+ if value.tzinfo is None:
+ return value.replace(tzinfo=timezone.utc)
+ return value.astimezone(timezone.utc)
+
+
+def _retry_after_seconds(locked_until: Optional[datetime], now: datetime) -> Optional[int]:
+ if not locked_until:
+ return None
+ locked_until = _utc_now(locked_until)
+ remaining = int((locked_until - now).total_seconds())
+ return max(1, remaining) if remaining > 0 else None
+
+
+async def get_throttle_state(
+ session: AsyncSession,
+ *,
+ scope: str,
+ identifier: str,
+) -> Optional[SecurityThrottle]:
+ stmt = (
+ select(SecurityThrottle)
+ .where(
+ SecurityThrottle.scope == scope,
+ SecurityThrottle.identifier == identifier,
+ )
+ .limit(1)
+ )
+ result = await session.execute(stmt)
+ return result.scalar_one_or_none()
+
+
+async def check_throttle(
+ session: AsyncSession,
+ *,
+ scope: str,
+ identifier: str,
+ now: Optional[datetime] = None,
+) -> ThrottleDecision:
+ now = _utc_now(now)
+ row = await get_throttle_state(session, scope=scope, identifier=identifier)
+ if not row or not row.locked_until:
+ return ThrottleDecision(locked=False)
+
+ locked_until = _utc_now(row.locked_until)
+ if locked_until <= now:
+ return ThrottleDecision(locked=False)
+
+ return ThrottleDecision(
+ locked=True,
+ retry_after=_retry_after_seconds(locked_until, now),
+ )
+
+
+async def record_throttle_failure(
+ session: AsyncSession,
+ *,
+ scope: str,
+ identifier: str,
+ max_failures: int,
+ window_seconds: int,
+ lock_seconds: int,
+ now: Optional[datetime] = None,
+) -> ThrottleDecision:
+ now = _utc_now(now)
+ max_failures = max(1, int(max_failures))
+ window_seconds = max(1, int(window_seconds))
+ lock_seconds = max(1, int(lock_seconds))
+ window_cutoff = now - timedelta(seconds=window_seconds)
+ lock_until = now + timedelta(seconds=lock_seconds)
+
+ failure_count_expr = case(
+ (
+ or_(
+ SecurityThrottle.window_started_at.is_(None),
+ SecurityThrottle.window_started_at <= window_cutoff,
+ ),
+ 1,
+ ),
+ else_=SecurityThrottle.failures + 1,
+ )
+
+ stmt = (
+ pg_insert(SecurityThrottle)
+ .values(
+ scope=scope,
+ identifier=identifier,
+ failures=1,
+ window_started_at=now,
+ last_attempt_at=now,
+ locked_until=lock_until if max_failures <= 1 else None,
+ )
+ .on_conflict_do_update(
+ index_elements=[SecurityThrottle.scope, SecurityThrottle.identifier],
+ set_={
+ "failures": failure_count_expr,
+ "window_started_at": case(
+ (
+ or_(
+ SecurityThrottle.window_started_at.is_(None),
+ SecurityThrottle.window_started_at <= window_cutoff,
+ ),
+ now,
+ ),
+ else_=SecurityThrottle.window_started_at,
+ ),
+ "last_attempt_at": now,
+ "locked_until": case(
+ (failure_count_expr >= max_failures, lock_until),
+ else_=None,
+ ),
+ },
+ )
+ .returning(SecurityThrottle.locked_until)
+ )
+
+ result = await session.execute(stmt)
+ locked_until = result.scalar_one_or_none()
+ locked_until = _utc_now(locked_until) if locked_until else None
+ if locked_until and locked_until > now:
+ return ThrottleDecision(
+ locked=True,
+ retry_after=_retry_after_seconds(locked_until, now),
+ )
+ return ThrottleDecision(locked=False)
+
+
+async def clear_throttle_state(
+ session: AsyncSession,
+ *,
+ scope: str,
+ identifier: str,
+) -> None:
+ stmt = delete(SecurityThrottle).where(
+ SecurityThrottle.scope == scope,
+ SecurityThrottle.identifier == identifier,
+ )
+ await session.execute(stmt)
diff --git a/db/migrator.py b/db/migrator.py
index f4ce9b1..e0062f2 100644
--- a/db/migrator.py
+++ b/db/migrator.py
@@ -202,6 +202,43 @@ def _migration_0005_add_email_auth_fields(connection: Connection) -> None:
)
)
+
+def _migration_0006_add_security_throttles(connection: Connection) -> None:
+ connection.execute(
+ text(
+ """
+ CREATE TABLE IF NOT EXISTS security_throttles (
+ throttle_id SERIAL PRIMARY KEY,
+ scope VARCHAR(64) NOT NULL,
+ identifier VARCHAR(512) NOT NULL,
+ failures INTEGER NOT NULL DEFAULT 0,
+ window_started_at TIMESTAMPTZ NULL,
+ locked_until TIMESTAMPTZ NULL,
+ last_attempt_at TIMESTAMPTZ NULL,
+ created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+ updated_at TIMESTAMPTZ NULL,
+ CONSTRAINT uq_security_throttles_scope_identifier UNIQUE (scope, identifier)
+ )
+ """
+ )
+ )
+ connection.execute(
+ text(
+ """
+ CREATE INDEX IF NOT EXISTS ix_security_throttles_scope
+ ON security_throttles (scope)
+ """
+ )
+ )
+ connection.execute(
+ text(
+ """
+ CREATE INDEX IF NOT EXISTS ix_security_throttles_locked_until
+ ON security_throttles (locked_until)
+ """
+ )
+ )
+
MIGRATIONS: List[Migration] = [
Migration(
id="0001_add_channel_subscription_fields",
@@ -228,6 +265,11 @@ MIGRATIONS: List[Migration] = [
description="Add email login identities and verification codes",
upgrade=_migration_0005_add_email_auth_fields,
),
+ Migration(
+ id="0006_add_security_throttles",
+ description="Add generic lockout tracking for brute-force protection",
+ upgrade=_migration_0006_add_security_throttles,
+ ),
]
diff --git a/db/models.py b/db/models.py
index 78a3793..9b7dda9 100644
--- a/db/models.py
+++ b/db/models.py
@@ -110,6 +110,24 @@ class EmailVerificationCode(Base):
target_user = relationship("User")
+class SecurityThrottle(Base):
+ __tablename__ = "security_throttles"
+
+ throttle_id = Column(Integer, primary_key=True, autoincrement=True)
+ scope = Column(String(64), nullable=False, index=True)
+ identifier = Column(String(512), nullable=False, index=True)
+ failures = Column(Integer, nullable=False, default=0)
+ window_started_at = Column(DateTime(timezone=True), nullable=True)
+ locked_until = Column(DateTime(timezone=True), nullable=True, index=True)
+ last_attempt_at = Column(DateTime(timezone=True), nullable=True)
+ created_at = Column(DateTime(timezone=True), server_default=func.now())
+ updated_at = Column(DateTime(timezone=True), onupdate=func.now(), nullable=True)
+
+ __table_args__ = (
+ UniqueConstraint("scope", "identifier", name="uq_security_throttles_scope_identifier"),
+ )
+
+
class Payment(Base):
__tablename__ = "payments"
diff --git a/locales/en.json b/locales/en.json
index d53c201..94d6bc8 100644
--- a/locales/en.json
+++ b/locales/en.json
@@ -87,6 +87,7 @@
"traffic_used_with_period": "{traffic_used} ({traffic_period})",
"promo_code_prompt": "Please enter your promo code:",
"promo_code_not_found": "Promo code {code} not found, expired, or already used the maximum number of times.",
+ "promo_code_too_many_attempts": "Too many failed promo code attempts. Please try again in {seconds} sec.",
"promo_code_already_used_by_user": "You have already used promo code {code}.",
"promo_code_applied_success_full": "✅ Promo code applied successfully!\nSubscription active until {end_date}.\n\nConnection key:\n{config_link}\n\nTo connect, open the link and follow the instructions 👇",
"error_applying_promo_bonus": "Failed to apply promo bonus. Please try again later or contact support.",
diff --git a/locales/ru.json b/locales/ru.json
index 59e881c..1f8445f 100644
--- a/locales/ru.json
+++ b/locales/ru.json
@@ -87,6 +87,7 @@
"traffic_used_with_period": "{traffic_used} ({traffic_period})",
"promo_code_prompt": "Пожалуйста, введите ваш промокод:",
"promo_code_not_found": "Промокод {code} не найден, истек или уже использован максимальное количество раз.",
+ "promo_code_too_many_attempts": "Слишком много неудачных попыток ввода промокода. Повторите через {seconds} сек.",
"promo_code_already_used_by_user": "Вы уже активировали промокод {code}.",
"promo_code_applied_success_full": "✅ Промокод успешно применен!\nПодписка активна до {end_date}.\n\nКлюч подключения:\n{config_link}\n\nЧтобы подключиться, перейдите по ссылке и следуйте инструкции 👇",
"error_applying_promo_bonus": "Не удалось применить бонус по промокоду. Пожалуйста, попробуйте позже или свяжитесь с поддержкой.",