houseassassin/remnawave-bedolaga-telegram-bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #859 from Fr1ngg/lxh4tz-bedolaga/add-external-admin-panel-settings

Browse Source 
Add external admin token support
This commit is contained in:
Egor
2025-10-08 01:42:46 +03:00
committed by GitHub
parent 6735cf17cf 83b370669e
commit cf76ac7893
6 changed files with 209 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/config.py
+18 -3
View File
@@ -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,
app/database/universal_migration.py
+26
View File
@@ -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:
app/handlers/admin/bot_configuration.py
+16 -14
View File
@@ -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="⬅️ Назад",
app/services/external_admin_token_service.py
+106
View File
@@ -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()
app/services/system_settings_service.py
+23 -1
View File
@@ -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)
main.py
+20
View File
@@ -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)