diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..8b54b37 --- /dev/null +++ b/.env.example @@ -0,0 +1,16 @@ +# Telegram Bot Token +# Получите токен у @BotFather в Telegram +TELEGRAM_BOT_TOKEN=123456789:ABCdefGHIjklMNOpqrsTUVwxyz + +# Настройки бота +BOT_NAME=CLI Assistant +BOT_DESCRIPTION=Бот для выполнения CLI команд +BOT_ICON_EMOJI=🤖 + +# Разрешённые пользователи (список ID через запятую) +# Пустой список = доступ открыт для всех +# Ваш ID можно узнать через @userinfobot +ALLOWED_USERS= + +# Рабочая директория для команд +WORKING_DIRECTORY=/home/mirivlad diff --git a/README.md b/README.md index 74b3613..3af3385 100644 --- a/README.md +++ b/README.md @@ -41,11 +41,37 @@ pip install -r requirements.txt 3. Следуйте инструкциям 4. Скопируйте полученный токен -### 5. Запуск бота +### 5. Настройка токена + +**Способ 1: Через файл .env (рекомендуется)** + +Скопируйте `.env.example` в `.env` и укажите токен: + +```bash +cp .env.example .env +# Отредактируйте .env, вставив ваш токен +``` + +**Способ 2: Через переменную окружения** ```bash export TELEGRAM_BOT_TOKEN='your_token_here' +``` + +**Способ 3: Интерактивная настройка** + +Запустите скрипт `run.sh` — он сам запросит токен: + +```bash +./run.sh +``` + +### 6. Запуск бота + +```bash python bot.py +# или через скрипт +./run.sh ``` ## Использование @@ -139,27 +165,35 @@ async def my_custom_command(update, context): ## Конфигурация -Настройки хранятся в `bot_config.json`: +Все настройки хранятся в файле `.env`: -```json -{ - "bot_name": "CLI Assistant", - "bot_description": "Бот для выполнения CLI команд", - "bot_icon_emoji": "🤖", - "allowed_users": [], - "require_confirmation": true, - "working_directory": "/home/user" -} +```bash +# Токен бота +TELEGRAM_BOT_TOKEN=123456789:ABCdefGHIjklMNOpqrsTUVwxyz + +# Настройки бота +BOT_NAME=CLI Assistant +BOT_DESCRIPTION=Бот для выполнения CLI команд +BOT_ICON_EMOJI=🤖 + +# Разрешённые пользователи (список ID через запятую) +# Пустой список = доступ открыт для всех +ALLOWED_USERS=123456789,987654321 + +# Рабочая директория для команд +WORKING_DIRECTORY=/home/user ``` | Параметр | Описание | |----------|----------| -| `bot_name` | Имя бота | -| `bot_description` | Описание бота | -| `bot_icon_emoji` | Emoji-иконка | -| `allowed_users` | Список разрешённых user ID (пусто = все) | -| `require_confirmation` | Требовать подтверждение перед выполнением | -| `working_directory` | Рабочая директория для команд | +| `TELEGRAM_BOT_TOKEN` | Токен бота от @BotFather | +| `BOT_NAME` | Отображаемое имя бота | +| `BOT_DESCRIPTION` | Описание бота | +| `BOT_ICON_EMOJI` | Emoji-иконка | +| `ALLOWED_USERS` | Список разрешённых user ID через запятую (пусто = все) | +| `WORKING_DIRECTORY` | Рабочая директория для выполнения команд | + +⚠️ **Важно:** После изменения `.env` требуется перезапуск бота. ## Безопасность @@ -167,8 +201,14 @@ async def my_custom_command(update, context): 1. Бот выполняет команды от имени запустившего пользователя 2. Не запускайте бота от root -3. Ограничьте доступ через `allowed_users` +3. Ограничьте доступ через `ALLOWED_USERS` в `.env`: + ```bash + ALLOWED_USERS=123456789,987654321 + ``` + Ваш ID можно узнать через @userinfobot 4. Будьте осторожны с деструктивными командами (`rm`, `dd`, etc.) +5. **Никогда не передавайте файл `.env`** — он содержит токен бота +6. Добавьте `.env` в `.gitignore` (уже сделано) ## Логи @@ -180,10 +220,12 @@ async def my_custom_command(update, context): telegram-cli-bot/ ├── bot.py # Основной файл бота ├── requirements.txt # Зависимости Python -├── bot_config.json # Конфигурация (создаётся автоматически) -├── bot.log # Лог файл -├── .gitignore # Git ignore -└── README.md # Документация +├── .env # Конфигурация (создаётся автоматически, не коммитить) +├── .env.example # Пример конфигурации +├── .gitignore # Git ignore +├── bot.log # Лог файл +├── run.sh # Скрипт запуска +└── README.md # Документация ``` ## Требования diff --git a/bot.py b/bot.py index c4739f9..3bd11e1 100644 --- a/bot.py +++ b/bot.py @@ -6,14 +6,15 @@ Telegram CLI Bot - бот для выполнения CLI команд с мно import os import sys -import json import asyncio import subprocess import logging from pathlib import Path from typing import Optional, Callable, Dict, Any, List from dataclasses import dataclass, field +from functools import wraps +from dotenv import load_dotenv from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup, BotCommand from telegram.ext import ( Application, @@ -24,10 +25,11 @@ from telegram.ext import ( filters, ) +# Загрузка переменных окружения из .env +load_dotenv() + # --- Конфигурация --- BASE_DIR = Path(__file__).parent -CONFIG_FILE = BASE_DIR / "bot_config.json" -COMMANDS_FILE = BASE_DIR / "commands.yaml" logging.basicConfig( format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", @@ -40,80 +42,60 @@ logging.basicConfig( logger = logging.getLogger(__name__) +# --- Конфигурация бота из переменных окружения --- +class BotConfig: + """Конфигурация бота из переменных окружения.""" + + def __init__(self): + self.name = os.getenv("BOT_NAME", "CLI Assistant") + self.description = os.getenv("BOT_DESCRIPTION", "Бот для выполнения CLI команд") + self.icon = os.getenv("BOT_ICON_EMOJI", "🤖") + self.working_directory = os.getenv("WORKING_DIRECTORY", str(Path.home())) + + # Парсинг списка разрешённых пользователей + allowed_users_str = os.getenv("ALLOWED_USERS", "") + if allowed_users_str.strip(): + self.allowed_users = [ + int(uid.strip()) + for uid in allowed_users_str.split(",") + if uid.strip().isdigit() + ] + else: + self.allowed_users = [] + + @property + def is_access_restricted(self) -> bool: + """Проверка: ограничен ли доступ.""" + return len(self.allowed_users) > 0 + + # --- Хранилище состояний пользователя --- @dataclass class UserState: """Состояние пользователя в диалоге.""" current_menu: str = "main" waiting_for_input: bool = False - input_type: Optional[str] = None # "name", "description", "icon", "command" + input_type: Optional[str] = None parent_menu: Optional[str] = None context: Dict[str, Any] = field(default_factory=dict) - working_directory: Optional[str] = None # Текущая директория пользователя + working_directory: Optional[str] = None class StateManager: """Управление состояниями пользователей.""" - + def __init__(self): self._states: Dict[int, UserState] = {} - + def get(self, user_id: int) -> UserState: if user_id not in self._states: self._states[user_id] = UserState() return self._states[user_id] - + def reset(self, user_id: int): self._states[user_id] = UserState() -# --- Конфигурация бота --- -class BotConfig: - """Конфигурация бота с сохранением в JSON.""" - - DEFAULT_CONFIG = { - "bot_name": "CLI Assistant", - "bot_description": "Бот для выполнения CLI команд", - "bot_icon_emoji": "🤖", - "allowed_users": [], # пустой список = все разрешены - "require_confirmation": True, - "working_directory": str(Path.home()), - } - - def __init__(self, config_file: Path = CONFIG_FILE): - self.config_file = config_file - self._config = self._load() - - def _load(self) -> dict: - if self.config_file.exists(): - with open(self.config_file, "r", encoding="utf-8") as f: - return json.load(f) - return self.DEFAULT_CONFIG.copy() - - def _save(self): - with open(self.config_file, "w", encoding="utf-8") as f: - json.dump(self._config, f, indent=2, ensure_ascii=False) - - def get(self, key: str, default=None): - return self._config.get(key, self.DEFAULT_CONFIG.get(key, default)) - - def set(self, key: str, value): - self._config[key] = value - self._save() - - @property - def name(self) -> str: - return self._config.get("bot_name", self.DEFAULT_CONFIG["bot_name"]) - - @property - def description(self) -> str: - return self._config.get("bot_description", self.DEFAULT_CONFIG["bot_description"]) - - @property - def icon(self) -> str: - return self._config.get("bot_icon_emoji", self.DEFAULT_CONFIG["bot_icon_emoji"]) - - # --- Система команд --- @dataclass class MenuItem: @@ -180,6 +162,31 @@ menu_builder = MenuBuilder() command_registry = CommandRegistry() +# --- Проверка прав доступа --- +def check_access(func): + """Декоратор для проверки прав доступа пользователя.""" + @wraps(func) + async def wrapper(update: Update, context: ContextTypes.DEFAULT_TYPE, *args, **kwargs): + user_id = update.effective_user.id + + # Если доступ не ограничен — пропускаем всех + if not config.is_access_restricted: + return await func(update, context, *args, **kwargs) + + if user_id not in config.allowed_users: + logger.warning(f"Попытка доступа от запрещённого пользователя {user_id}") + await update.message.reply_text( + "❌ *Доступ запрещён*\n\n" + "Ваш ID не добавлен в список разрешённых пользователей.\n" + f"Ваш ID: `{user_id}`", + parse_mode="Markdown" + ) + return + + return await func(update, context, *args, **kwargs) + return wrapper + + # --- Инициализация меню --- def init_menus(): """Инициализация структуры меню.""" @@ -264,15 +271,16 @@ def init_menus(): # --- Хендлеры --- +@check_access async def start_command(update: Update, context: ContextTypes.DEFAULT_TYPE): """Обработка команды /start.""" user = update.effective_user logger.info(f"Пользователь {user.username} ({user.id}) запустил бота") - + state_manager.reset(user.id) - + # Показать текущую директорию - working_dir = config.get("working_directory", str(Path.home())) + working_dir = config.working_directory await update.message.reply_text( f"👋 Привет, {user.first_name}!\n\n" @@ -288,17 +296,18 @@ async def start_command(update: Update, context: ContextTypes.DEFAULT_TYPE): ) +@check_access async def menu_command(update: Update, context: ContextTypes.DEFAULT_TYPE): """Обработка команды /menu - показывает главное меню.""" user = update.effective_user state = state_manager.get(user.id) - + # Сброс состояния и возврат к главному меню state_manager.reset(user.id) state.current_menu = "main" - + # Показать текущую директорию - working_dir = state.working_directory or config.get("working_directory", str(Path.home())) + working_dir = state.working_directory or config.working_directory await update.message.reply_text( f"🏠 *Главное меню*\n\n" @@ -309,6 +318,7 @@ async def menu_command(update: Update, context: ContextTypes.DEFAULT_TYPE): ) +@check_access async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE): """Обработка команды /help.""" help_text = f""" @@ -347,6 +357,7 @@ async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE): await update.message.reply_text(help_text, parse_mode="Markdown") +@check_access async def settings_command(update: Update, context: ContextTypes.DEFAULT_TYPE): """Обработка команды /settings.""" state = state_manager.get(update.effective_user.id) @@ -445,65 +456,57 @@ async def menu_callback(update: Update, context: ContextTypes.DEFAULT_TYPE): else: await query.edit_message_text("❌ Команда не найдена") - # Настройки бота + # Настройки бота - только просмотр, изменение через .env elif callback == "set_name": - state.waiting_for_input = True - state.input_type = "name" await query.edit_message_text( "📝 *Изменение имени бота*\n\n" f"Текущее имя: `{config.name}`\n\n" - "Отправьте новое имя бота:", + "Для изменения отредактируйте `.env`:\n" + "```\nBOT_NAME=Ваше имя\n```", parse_mode="Markdown", reply_markup=menu_builder.get_keyboard("settings") ) - + elif callback == "set_description": - state.waiting_for_input = True - state.input_type = "description" await query.edit_message_text( "📄 *Изменение описания бота*\n\n" f"Текущее описание: `{config.description}`\n\n" - "Отправьте новое описание:", + "Для изменения отредактируйте `.env`:\n" + "```\nBOT_DESCRIPTION=Ваше описание\n```", parse_mode="Markdown", reply_markup=menu_builder.get_keyboard("settings") ) - + elif callback == "set_icon": - state.waiting_for_input = True - state.input_type = "icon" await query.edit_message_text( "🎨 *Изменение иконки бота*\n\n" f"Текущая иконка: `{config.icon}`\n\n" - "Отправьте новый emoji (один символ):", + "Для изменения отредактируйте `.env`:\n" + "```\nBOT_ICON_EMOJI=🤖\n```", parse_mode="Markdown", reply_markup=menu_builder.get_keyboard("settings") ) - + elif callback == "show_access": - allowed = config.get("allowed_users", []) - if allowed: - text = "👥 *Разрешённые пользователи:*\n" + "\n".join(f"• `{uid}`" for uid in allowed) + if config.allowed_users: + text = "👥 *Разрешённые пользователи:*\n" + "\n".join(f"• `{uid}`" for uid in config.allowed_users) else: text = "👥 *Доступ открыт для всех*\n\n(список разрешённых пользователей пуст)" await query.edit_message_text(text, parse_mode="Markdown") - + elif callback == "add_access": - state.waiting_for_input = True - state.input_type = "add_access" await query.edit_message_text( "➕ *Добавление пользователя*\n\n" - "Отправьте ID пользователя Telegram:\n" - "(можно получить через @userinfobot)", + "Для добавления пользователя отредактируйте `.env`:\n" + "```\nALLOWED_USERS=123456789,987654321\n```\n" + "Ваш ID можно узнать через @userinfobot", parse_mode="Markdown" ) - + elif callback == "remove_access": - state.waiting_for_input = True - state.input_type = "remove_access" - allowed = config.get("allowed_users", []) - if allowed: - text = "➖ *Удаление пользователя*\n\n" + "\n".join(f"• `{uid}`" for uid in allowed) - text += "\n\nОтправьте ID для удаления:" + if config.allowed_users: + text = "➖ *Удаление пользователя*\n\n" + "\n".join(f"• `{uid}`" for uid in config.allowed_users) + text += "\n\nУдалите ID из `.env` чтобы убрать доступ" else: text = "➖ Список пуст, некого удалять" await query.edit_message_text(text, parse_mode="Markdown") @@ -514,7 +517,7 @@ async def menu_callback(update: Update, context: ContextTypes.DEFAULT_TYPE): f"*{config.icon} {config.name}*\n" f"_{config.description}_\n\n" f"Версия: 1.0.0\n" - f"Рабочая директория: `{config.get('working_directory')}`\n\n" + f"Рабочая директория: `{config.working_directory}`\n\n" f"Бот позволяет выполнять CLI команды на вашем ПК\n" f"через интерфейс Telegram.", parse_mode="Markdown", @@ -527,9 +530,9 @@ async def execute_cli_command(query, command: str): """Выполнение CLI команды из кнопки меню.""" user_id = query.from_user.id state = state_manager.get(user_id) - + # Определяем рабочую директорию: сначала пользовательская, потом из конфига - working_dir = state.working_directory or config.get("working_directory", str(Path.home())) + working_dir = state.working_directory or config.working_directory logger.info(f"Выполнение команды: {command} в директории: {working_dir}") @@ -585,78 +588,13 @@ async def execute_cli_command(query, command: str): ) -async def handle_settings_input(update: Update, context: ContextTypes.DEFAULT_TYPE, text: str): - """Обработка ввода в режиме настройки.""" - user_id = update.effective_user.id - state = state_manager.get(user_id) - input_type = state.input_type - - if input_type == "name": - config.set("bot_name", text) - await update.message.reply_text( - f"✅ Имя бота изменено на: `{text}`\n\n" - f"Используйте /start для возврата в главное меню", - parse_mode="Markdown" - ) - - elif input_type == "description": - config.set("bot_description", text) - await update.message.reply_text( - f"✅ Описание изменено на: `{text}`\n\n" - f"Используйте /start для возврата в главное меню", - parse_mode="Markdown" - ) - - elif input_type == "icon": - config.set("bot_icon_emoji", text[0] if text else "🤖") - await update.message.reply_text( - f"✅ Иконка изменена на: `{text[0] if text else '🤖'}`\n\n" - f"Используйте /start для возврата в главное меню", - parse_mode="Markdown" - ) - - elif input_type == "add_access": - try: - uid = int(text) - allowed = config.get("allowed_users", []) - if uid not in allowed: - allowed.append(uid) - config.set("allowed_users", allowed) - await update.message.reply_text(f"✅ Пользователь `{uid}` добавлен", parse_mode="Markdown") - else: - await update.message.reply_text(f"⚠️ Пользователь `{uid}` уже в списке", parse_mode="Markdown") - except ValueError: - await update.message.reply_text("❌ Неверный формат ID") - - elif input_type == "remove_access": - try: - uid = int(text) - allowed = config.get("allowed_users", []) - if uid in allowed: - allowed.remove(uid) - config.set("allowed_users", allowed) - await update.message.reply_text(f"✅ Пользователь `{uid}` удалён", parse_mode="Markdown") - else: - await update.message.reply_text(f"⚠️ Пользователь `{uid}` не найден", parse_mode="Markdown") - except ValueError: - await update.message.reply_text("❌ Неверный формат ID") - - # Сброс состояния - state.waiting_for_input = False - state.input_type = None - - +@check_access async def handle_text_message(update: Update, context: ContextTypes.DEFAULT_TYPE): """Обработка текстовых сообщений как CLI команд.""" user_id = update.effective_user.id text = update.message.text.strip() state = state_manager.get(user_id) - - # Проверка: не в режиме настройки ли мы - if state.waiting_for_input: - await handle_settings_input(update, context, text) - return - + # Любое текстовое сообщение = CLI команда logger.info(f"Пользователь {user_id} отправил команду: {text}") @@ -671,9 +609,9 @@ async def execute_cli_command_from_message(update: Update, command: str): """Выполнение CLI команды из сообщения.""" user_id = update.effective_user.id state = state_manager.get(user_id) - + # Определяем рабочую директорию: сначала пользовательская, потом из конфига - working_dir = state.working_directory or config.get("working_directory", str(Path.home())) + working_dir = state.working_directory or config.working_directory # Обработка команды cd - меняем директорию пользователя # Работает только с простыми командами cd, не с составными @@ -829,33 +767,21 @@ async def post_init(application: Application): ] await application.bot.set_my_commands(commands) - # Установка имени и описания - await application.bot.set_my_name(config.name) - await application.bot.set_my_description(config.description) - logger.info("Бот инициализирован") def main(): """Точка входа.""" - # Проверка токена: сначала переменная окружения, потом конфиг + # Чтение токена только из переменной окружения token = os.getenv("TELEGRAM_BOT_TOKEN") - - if not token and CONFIG_FILE.exists(): - try: - with open(CONFIG_FILE, "r", encoding="utf-8") as f: - config_data = json.load(f) - token = config_data.get("bot_token") - if token: - logger.info("Токен получен из конфигурации") - except Exception as e: - logger.warning(f"Не удалось прочитать токен из конфига: {e}") - + if not token: print("❌ Ошибка: не установлен TELEGRAM_BOT_TOKEN") - print("Задайте переменную окружения:") - print(" export TELEGRAM_BOT_TOKEN='your_token_here'") - print("Или запустите ./run.sh для интерактивной настройки") + print("\nСпособы установки токена:") + print(" 1. Создайте файл .env по примеру .env.example") + print(" 2. Или задайте переменную окружения:") + print(" export TELEGRAM_BOT_TOKEN='your_token_here'") + print("\nИли запустите ./run.sh для интерактивной настройки") sys.exit(1) # Инициализация меню diff --git a/requirements.txt b/requirements.txt index 92aa08a..a52f235 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,2 +1,3 @@ python-telegram-bot==21.0 pyyaml==6.0.1 +python-dotenv==1.0.1 diff --git a/run.sh b/run.sh index d145bb8..d5f2022 100755 --- a/run.sh +++ b/run.sh @@ -6,23 +6,25 @@ set -e SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" cd "$SCRIPT_DIR" -CONFIG_FILE="$SCRIPT_DIR/bot_config.json" +ENV_FILE="$SCRIPT_DIR/.env" -# Функция для получения значения из JSON -get_json_value() { - python3 -c "import json; data=json.load(open('$CONFIG_FILE')); print(data.get('$1', ''))" 2>/dev/null || echo "" -} - -# Функция для установки значения в JSON -set_json_value() { - python3 -c " -import json -with open('$CONFIG_FILE', 'r') as f: - data = json.load(f) -data['$1'] = '$2' -with open('$CONFIG_FILE', 'w') as f: - json.dump(data, f, indent=2, ensure_ascii=False) -" +# Функция для установки значения в .env +set_env_value() { + local key="$1" + local value="$2" + + if [ -f "$ENV_FILE" ]; then + # Если ключ существует - обновляем + if grep -q "^$key=" "$ENV_FILE"; then + sed -i "s|^$key=.*|$key=$value|" "$ENV_FILE" + else + # Иначе добавляем + echo "$key=$value" >> "$ENV_FILE" + fi + else + # Создаём файл + echo "$key=$value" > "$ENV_FILE" + fi } # Проверка виртуального окружения @@ -49,20 +51,20 @@ pip install -q -r requirements.txt # Работа с токеном TOKEN="" -# 1. Проверяем переменную окружения +# 1. Проверяем .env файл +if [ -f "$ENV_FILE" ]; then + TOKEN=$(grep "^TELEGRAM_BOT_TOKEN=" "$ENV_FILE" | cut -d'=' -f2) + if [ -n "$TOKEN" ]; then + echo "✅ Токен получен из .env" + fi +fi + +# 2. Проверяем переменную окружения (имеет приоритет над .env) if [ -n "$TELEGRAM_BOT_TOKEN" ]; then TOKEN="$TELEGRAM_BOT_TOKEN" echo "✅ Токен получен из переменной окружения" fi -# 2. Если нет в переменной, проверяем конфиг -if [ -z "$TOKEN" ] && [ -f "$CONFIG_FILE" ]; then - TOKEN=$(get_json_value "bot_token") - if [ -n "$TOKEN" ]; then - echo "✅ Токен получен из конфигурации" - fi -fi - # 3. Если токена нет нигде, запрашиваем у пользователя if [ -z "$TOKEN" ]; then echo "" @@ -81,22 +83,18 @@ if [ -z "$TOKEN" ]; then echo "" echo "━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━" echo "" - + # Запрос токена с подтверждением while true; do read -p "📋 Вставьте токен бота: " TOKEN - + # Проверка формата токена (примерно 46 символов, содержит : и _) if [[ "$TOKEN" =~ ^[0-9]+:[A-Za-z0-9_-]+$ ]]; then echo "" - read -p "💾 Сохранить токен в конфигурацию? (y/n): " SAVE + read -p "💾 Сохранить токен в .env? (y/n): " SAVE if [[ "$SAVE" =~ ^[Yy]$ ]]; then - # Создаём конфиг если нет - if [ ! -f "$CONFIG_FILE" ]; then - echo '{}' > "$CONFIG_FILE" - fi - set_json_value "bot_token" "$TOKEN" - echo "✅ Токен сохранён в $CONFIG_FILE" + set_env_value "TELEGRAM_BOT_TOKEN" "$TOKEN" + echo "✅ Токен сохранён в $ENV_FILE" fi break else