import logging import re from datetime import datetime from typing import Optional from aiogram import Bot, F, Router, types from aiogram.fsm.context import FSMContext from sqlalchemy.ext.asyncio import AsyncSession from bot.keyboards.inline.user_keyboards import ( get_back_to_main_menu_markup, get_connect_and_main_keyboard, ) from bot.middlewares.i18n import JsonI18n from bot.services.promo_code_service import PromoCodeService from bot.services.subscription_service import SubscriptionService from bot.states.user_states import UserPromoStates from bot.utils.callback_answer import safe_answer_callback from bot.utils.install_links import ( append_install_share_link_text, ensure_user_install_guide_links, ) from config.settings import Settings from db.dal import user_dal from .start import send_main_menu router = Router(name="user_promo_router") SUSPICIOUS_SQL_KEYWORDS_REGEX = re.compile( r"\b(DROP\s*TABLE|DELETE\s*FROM|ALTER\s*TABLE|TRUNCATE\s*TABLE|UNION\s*SELECT|" r";\s*SELECT|;\s*INSERT|;\s*UPDATE|;\s*DELETE|xp_cmdshell|sysdatabases|sysobjects|INFORMATION_SCHEMA)\b", re.IGNORECASE, ) SUSPICIOUS_CHARS_REGEX = re.compile(r"(--|#\s|;|\*\/|\/\*)") MAX_PROMO_CODE_INPUT_LENGTH = 100 async def prompt_promo_code_input( callback: types.CallbackQuery, state: FSMContext, i18n_data: dict, settings: Settings, session: AsyncSession, back_callback: str = "main_action:back_to_main", ): current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE) i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance") if not i18n: await safe_answer_callback(callback, "Language service error.", show_alert=True) return _ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs) if not callback.message: logging.error("CallbackQuery has no message in prompt_promo_code_input") await safe_answer_callback( callback, _("error_occurred_processing_request"), show_alert=True, ) return try: await callback.message.edit_text( text=_(key="promo_code_prompt"), reply_markup=get_back_to_main_menu_markup( current_lang, i18n, callback_data=back_callback, ), ) except Exception as e_edit: logging.warning(f"Failed to edit message for promo prompt: {e_edit}. Sending new one.") await callback.message.answer( text=_(key="promo_code_prompt"), reply_markup=get_back_to_main_menu_markup( current_lang, i18n, callback_data=back_callback, ), ) await safe_answer_callback(callback) await state.set_state(UserPromoStates.waiting_for_promo_code) logging.info( f"User {callback.from_user.id} entered state UserPromoStates.waiting_for_promo_code. " f"FSM state: {await state.get_state()}" ) @router.message(UserPromoStates.waiting_for_promo_code, F.text) async def process_promo_code_input( message: types.Message, state: FSMContext, settings: Settings, i18n_data: dict, promo_code_service: PromoCodeService, subscription_service: SubscriptionService, bot: Bot, session: AsyncSession, ): logging.info( f"Processing promo code input from user {message.from_user.id} in state {await state.get_state()}: '{message.text}'" # noqa: E501 ) current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE) i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance") if not i18n or not promo_code_service: logging.error("Dependencies (i18n or PromoCodeService) missing in process_promo_code_input") await message.reply("Service error. Please try again later.") await state.clear() return _ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs) code_input = message.text.strip() if message.text else "" user = message.from_user is_suspicious = False if not code_input: is_suspicious = True logging.warning(f"Empty promo code input by user {user.id}.") elif ( len(code_input) > MAX_PROMO_CODE_INPUT_LENGTH or SUSPICIOUS_SQL_KEYWORDS_REGEX.search(code_input) or SUSPICIOUS_CHARS_REGEX.search(code_input) ): is_suspicious = True logging.warning( f"Suspicious input for promo code by user {user.id} (len: {len(code_input)}): '{code_input}'" # noqa: E501 ) response_to_user_text = "" if is_suspicious: # Send notification through NotificationService if enabled if settings.LOG_SUSPICIOUS_ACTIVITY: try: from bot.services.notification_service import NotificationService notification_service = NotificationService(bot, settings, i18n) db_user = await user_dal.get_user_by_id(session, user.id) await notification_service.notify_suspicious_promo_attempt( user_id=user.id, username=user.username, first_name=user.first_name, email=getattr(db_user, "email", None) if db_user else None, suspicious_input=code_input, ) except Exception as e: logging.error(f"Failed to send suspicious promo notification: {e}") 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, ) install_links = await ensure_user_install_guide_links(session, settings, user.id) install_share_url = install_links.public_share_url if install_share_url: try: await session.commit() response_to_user_text = append_install_share_link_text( response_to_user_text, _, install_share_url, ) except Exception: await session.rollback() logging.exception( "Failed to persist install guide share token for promo user %s.", user.id, ) install_share_url = None reply_markup = get_connect_and_main_keyboard( current_lang, i18n, settings, config_link_display, connect_button_url=connect_button_url, install_share_url=install_share_url, ) else: 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, reply_markup=reply_markup, parse_mode="HTML", ) await state.clear() logging.info( f"Promo code input '{code_input}' processing finished for user {message.from_user.id}. State cleared." # noqa: E501 ) @router.callback_query(F.data == "main_action:back_to_main", UserPromoStates.waiting_for_promo_code) async def cancel_promo_input_via_button( callback: types.CallbackQuery, state: FSMContext, settings: Settings, i18n_data: dict, subscription_service: SubscriptionService, session: AsyncSession, ): current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE) i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance") if not i18n: logging.error("i18n missing in cancel_promo_input_via_button") await safe_answer_callback(callback, "Language error", show_alert=True) return logging.info( f"User {callback.from_user.id} cancelled promo code input via button from state {await state.get_state()}. Clearing state." # noqa: E501 ) await state.clear() if callback.message: await send_main_menu( callback, settings, i18n_data, subscription_service, session, is_edit=True ) else: _ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs) await safe_answer_callback( callback, _("promo_input_cancelled_short"), show_alert=False, )