diff --git a/app/config.py b/app/config.py index 4d2f184a..450b27da 100644 --- a/app/config.py +++ b/app/config.py @@ -13,6 +13,7 @@ from pathlib import Path class Settings(BaseSettings): BOT_TOKEN: str + BOT_USERNAME: Optional[str] = None ADMIN_IDS: str = "" SUPPORT_USERNAME: str = "@support" SUPPORT_MENU_ENABLED: bool = True @@ -256,6 +257,8 @@ class Settings(BaseSettings): WEB_API_DEFAULT_TOKEN_NAME: str = "Bootstrap Token" WEB_API_TOKEN_HASH_ALGORITHM: str = "sha256" WEB_API_REQUEST_LOGGING: bool = True + + EXTERNAL_ADMIN_TOKEN: Optional[str] = None APP_CONFIG_PATH: str = "app-config.json" ENABLE_DEEP_LINKS: bool = True @@ -388,17 +391,29 @@ class Settings(BaseSettings): def get_admin_ids(self) -> List[int]: try: admin_ids = self.ADMIN_IDS - + if isinstance(admin_ids, str): if not admin_ids.strip(): return [] return [int(x.strip()) for x in admin_ids.split(',') if x.strip()] - + return [] - + except (ValueError, AttributeError): return [] + def get_bot_username(self) -> Optional[str]: + username = (self.BOT_USERNAME or "").strip() + if not username: + return None + return username.lstrip("@") or None + + def get_bot_username_display(self) -> Optional[str]: + username = self.get_bot_username() + if not username: + return None + return f"@{username}" + def get_remnawave_auth_params(self) -> Dict[str, Optional[str]]: return { "base_url": self.REMNAWAVE_API_URL, diff --git a/app/database/universal_migration.py b/app/database/universal_migration.py index 90d637ad..548be2cb 100644 --- a/app/database/universal_migration.py +++ b/app/database/universal_migration.py @@ -5,6 +5,7 @@ from sqlalchemy import inspect, select, text from sqlalchemy.ext.asyncio import AsyncSession from app.config import settings +from app.services.external_admin_token_service import external_admin_token_service from app.database.database import AsyncSessionLocal, engine from app.database.models import WebApiToken from app.utils.security import hash_api_token @@ -2769,6 +2770,22 @@ async def ensure_default_web_api_token() -> bool: return False +async def ensure_external_admin_token() -> bool: + token, updated = await external_admin_token_service.ensure_from_config() + + if token: + if updated: + logger.info("✅ Токен внешней админки создан или обновлен") + else: + logger.info("ℹ️ Токен внешней админки уже существует") + return True + + logger.warning( + "⚠️ Не удалось гарантировать токен внешней админки — укажите BOT_USERNAME" + ) + return False + + async def run_universal_migration(): logger.info("=== НАЧАЛО УНИВЕРСАЛЬНОЙ МИГРАЦИИ ===") @@ -2837,6 +2854,15 @@ async def run_universal_migration(): else: logger.warning("⚠️ Не удалось создать бутстрап токен веб-API") + logger.info("=== ГЕНЕРАЦИЯ ТОКЕНА ВНЕШНЕЙ АДМИНКИ ===") + external_admin_ready = await ensure_external_admin_token() + if external_admin_ready: + logger.info("✅ Токен внешней админки готов") + else: + logger.warning( + "⚠️ Токен внешней админки не создан — внешняя панель может быть недоступна" + ) + logger.info("=== СОЗДАНИЕ ТАБЛИЦЫ CRYPTOBOT ===") cryptobot_created = await create_cryptobot_payments_table() if cryptobot_created: diff --git a/app/handlers/admin/bot_configuration.py b/app/handlers/admin/bot_configuration.py index ad0bea38..bc5a0830 100644 --- a/app/handlers/admin/bot_configuration.py +++ b/app/handlers/admin/bot_configuration.py @@ -105,7 +105,7 @@ CATEGORY_GROUP_METADATA: Dict[str, Dict[str, object]] = { "title": "⚡ Расширенные", "description": "Web API, webhook, логирование и режим отладки.", "icon": "⚡", - "categories": ("WEB_API", "WEBHOOK", "LOG", "DEBUG"), + "categories": ("WEB_API", "EXTERNAL_ADMIN", "WEBHOOK", "LOG", "DEBUG"), }, } @@ -1361,6 +1361,7 @@ def _build_setting_keyboard( definition = bot_configuration_service.get_definition(key) rows: list[list[types.InlineKeyboardButton]] = [] callback_token = bot_configuration_service.get_callback_token(key) + readonly = bot_configuration_service.is_readonly(key) choice_options = bot_configuration_service.get_choice_options(key) if choice_options: @@ -1385,7 +1386,7 @@ def _build_setting_keyboard( for chunk in _chunk(choice_buttons, 2): rows.append(list(chunk)) - if definition.python_type is bool: + if definition.python_type is bool and not readonly: rows.append([ types.InlineKeyboardButton( text="🔁 Переключить", @@ -1395,25 +1396,26 @@ def _build_setting_keyboard( ) ]) - rows.append([ - types.InlineKeyboardButton( - text="✏️ Изменить", - callback_data=( - f"botcfg_edit:{group_key}:{category_page}:{settings_page}:{callback_token}" - ), - ) - ]) - - if bot_configuration_service.has_override(key): + if not readonly: rows.append([ types.InlineKeyboardButton( - text="♻️ Сбросить", + text="✏️ Изменить", callback_data=( - f"botcfg_reset:{group_key}:{category_page}:{settings_page}:{callback_token}" + f"botcfg_edit:{group_key}:{category_page}:{settings_page}:{callback_token}" ), ) ]) + if bot_configuration_service.has_override(key): + rows.append([ + types.InlineKeyboardButton( + text="♻️ Сбросить", + callback_data=( + f"botcfg_reset:{group_key}:{category_page}:{settings_page}:{callback_token}" + ), + ) + ]) + rows.append([ types.InlineKeyboardButton( text="⬅️ Назад", diff --git a/app/services/external_admin_token_service.py b/app/services/external_admin_token_service.py new file mode 100644 index 00000000..701b334c --- /dev/null +++ b/app/services/external_admin_token_service.py @@ -0,0 +1,106 @@ +import hashlib +import logging +from datetime import datetime +from typing import Optional, Tuple + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.config import settings +from app.database.database import AsyncSessionLocal +from app.database.models import SystemSetting + + +logger = logging.getLogger(__name__) + + +class ExternalAdminTokenService: + """Управляет токеном внешней административной панели.""" + + SETTING_KEY = "EXTERNAL_ADMIN_TOKEN" + SETTING_DESCRIPTION = "Токен для проверки внешней административной панели" + _SALT = "remnawave-external-admin-token" + + @staticmethod + def _normalize_username(username: Optional[str]) -> str: + if not username: + return "" + return username.strip().lstrip("@").lower() + + @staticmethod + def _sanitize_username(username: Optional[str]) -> Optional[str]: + if not username: + return None + sanitized = username.strip().lstrip("@") + return sanitized or None + + def generate_token(self, username: str) -> str: + normalized = self._normalize_username(username) + if not normalized: + raise ValueError("Bot username is required to generate external admin token") + payload = f"{self._SALT}:{normalized}".encode("utf-8") + digest = hashlib.sha256(payload).hexdigest() + return digest[:48] + + async def _ensure_in_session(self, db: AsyncSession, token: str) -> bool: + result = await db.execute( + select(SystemSetting).where(SystemSetting.key == self.SETTING_KEY) + ) + setting = result.scalar_one_or_none() + + if setting is None: + setting = SystemSetting( + key=self.SETTING_KEY, + value=token, + description=self.SETTING_DESCRIPTION, + ) + db.add(setting) + return True + + if setting.value != token: + setting.value = token + setting.description = setting.description or self.SETTING_DESCRIPTION + setting.updated_at = datetime.utcnow() + return True + + return False + + async def ensure_token( + self, + username: Optional[str], + *, + session: Optional[AsyncSession] = None, + ) -> Tuple[Optional[str], bool]: + """Гарантирует наличие токена для указанного username.""" + + normalized = self._normalize_username(username) + if not normalized: + logger.warning( + "Не удалось сгенерировать токен внешней админки: username не задан", + ) + return None, False + + token = self.generate_token(normalized) + sanitized_username = self._sanitize_username(username) + + if session is None: + async with AsyncSessionLocal() as db: + updated = await self._ensure_in_session(db, token) + await db.commit() + else: + updated = await self._ensure_in_session(session, token) + if updated: + await session.flush() + + settings.BOT_USERNAME = sanitized_username + settings.EXTERNAL_ADMIN_TOKEN = token + + return token, updated + + async def ensure_from_config(self) -> Tuple[Optional[str], bool]: + """Генерирует токен на основе username из конфигурации.""" + + return await self.ensure_token(settings.BOT_USERNAME) + + +external_admin_token_service = ExternalAdminTokenService() diff --git a/app/services/system_settings_service.py b/app/services/system_settings_service.py index e8529e0d..73418132 100644 --- a/app/services/system_settings_service.py +++ b/app/services/system_settings_service.py @@ -57,7 +57,8 @@ class ChoiceOption: class BotConfigurationService: - EXCLUDED_KEYS: set[str] = {"BOT_TOKEN", "ADMIN_IDS"} + EXCLUDED_KEYS: set[str] = {"BOT_TOKEN", "ADMIN_IDS", "BOT_USERNAME"} + READONLY_KEYS: set[str] = {"EXTERNAL_ADMIN_TOKEN"} CATEGORY_TITLES: Dict[str, str] = { "CORE": "🤖 Основные настройки", @@ -104,6 +105,7 @@ class BotConfigurationService: "WEBHOOK": "🌐 Webhook", "LOG": "📝 Логирование", "DEBUG": "🧪 Режим разработки", + "EXTERNAL_ADMIN": "🛡️ Внешняя админка", } CATEGORY_DESCRIPTIONS: Dict[str, str] = { @@ -151,6 +153,7 @@ class BotConfigurationService: "WEBHOOK": "Пути и секреты вебхуков.", "LOG": "Уровни логирования и ротация.", "DEBUG": "Отладочные функции и безопасный режим.", + "EXTERNAL_ADMIN": "Статический токен доступа для внешней административной панели.", } CATEGORY_KEY_OVERRIDES: Dict[str, str] = { @@ -231,6 +234,7 @@ class BotConfigurationService: "VERSION_CHECK_INTERVAL_HOURS": "VERSION", "TELEGRAM_STARS_RATE_RUB": "TELEGRAM", "REMNAWAVE_USER_DESCRIPTION_TEMPLATE": "REMNAWAVE", + "EXTERNAL_ADMIN_TOKEN": "EXTERNAL_ADMIN", } CATEGORY_PREFIX_OVERRIDES: Dict[str, str] = { @@ -269,6 +273,7 @@ class BotConfigurationService: "LOG_": "LOG", "WEB_API_": "WEB_API", "DEBUG": "DEBUG", + "EXTERNAL_ADMIN_": "EXTERNAL_ADMIN", } CHOICES: Dict[str, List[ChoiceOption]] = { @@ -394,6 +399,13 @@ class BotConfigurationService: "warning": "Недоступный адрес приведет к ошибкам при управлении VPN-учетками.", "dependencies": "REMNAWAVE_API_KEY или REMNAWAVE_USERNAME/REMNAWAVE_PASSWORD", }, + "EXTERNAL_ADMIN_TOKEN": { + "description": "Используется внешней админкой для проверки запросов к API.", + "format": "Значение формируется автоматически и доступно только для чтения.", + "example": "b5c8e1d4b7f0a93c4e2d1f6a8c7b5d1e", + "warning": "Изменение username бота приведет к изменению токена. Сохраните его в конфигурации внешней панели.", + "dependencies": "BOT_USERNAME", + }, } @classmethod @@ -405,6 +417,10 @@ class BotConfigurationService: definition = cls.get_definition(key) return definition.python_type is bool + @classmethod + def is_readonly(cls, key: str) -> bool: + return key in cls.READONLY_KEYS + @classmethod def _format_numeric_with_unit(cls, key: str, value: Union[int, float]) -> Optional[str]: if isinstance(value, bool): @@ -455,6 +471,8 @@ class BotConfigurationService: cleaned = value.strip() if not cleaned: return "—" + if key == "EXTERNAL_ADMIN_TOKEN": + return cleaned if any(keyword in key.upper() for keyword in ("TOKEN", "SECRET", "PASSWORD", "KEY")): return "••••••••" items = cls._split_comma_values(cleaned) @@ -861,6 +879,8 @@ class BotConfigurationService: @classmethod async def set_value(cls, db: AsyncSession, key: str, value: Any) -> None: + if cls.is_readonly(key): + raise ValueError("Эта настройка доступна только для чтения") raw_value = cls.serialize_value(key, value) await upsert_system_setting(db, key, raw_value) cls._overrides_raw[key] = raw_value @@ -871,6 +891,8 @@ class BotConfigurationService: @classmethod async def reset_value(cls, db: AsyncSession, key: str) -> None: + if cls.is_readonly(key): + raise ValueError("Эта настройка нельзя сбросить") await delete_system_setting(db, key) cls._overrides_raw.pop(key, None) original = cls.get_original_value(key) diff --git a/main.py b/main.py index 51d6bfa2..bff5638f 100644 --- a/main.py +++ b/main.py @@ -22,6 +22,7 @@ from app.services.backup_service import backup_service from app.services.reporting_service import reporting_service from app.localization.loader import ensure_locale_templates from app.services.system_settings_service import bot_configuration_service +from app.services.external_admin_token_service import external_admin_token_service from app.services.broadcast_service import broadcast_service from app.utils.startup_timeline import StartupTimeline @@ -133,6 +134,25 @@ async def main(): bot, dp = await setup_bot() stage.log("Кеш и FSM подготовлены") + try: + bot_profile = await bot.get_me() + token, updated = await external_admin_token_service.ensure_token( + bot_profile.username + ) + if token: + if updated: + await bot_configuration_service.reload() + stage.log("Токен внешней админки синхронизирован") + else: + stage.log("Токен внешней админки без изменений") + else: + stage.warning("Не удалось получить username бота для токена внешней админки") + except Exception as error: + stage.warning(f"Ошибка синхронизации токена внешней админки: {error}") + logging.getLogger(__name__).warning( + "Не удалось синхронизировать токен внешней админки: %s", error + ) + monitoring_service.bot = bot maintenance_service.set_bot(bot) broadcast_service.set_bot(bot)