Server control by Telegram Bot - Run Shell commands by Python

I have more than one Telegram bots for different purposes but one day I realise I need have a control of selected services on my server. There are several benefits of using a Telegram bot that has access to a Linux server terminal for services restart or status checking:

Remote access: With a Telegram bot, you can access your Linux server from anywhere in the world, as long as you have an internet connection. This means you can manage your server even if you’re not physically present.
Increased efficiency: Having a Telegram bot to manage your Linux server eliminates the need for you to log in to the server every time you need to check the status or restart a service. This can save you time and increase your overall efficiency.
Enhanced security: By using a Telegram bot for server management, you can avoid exposing your server’s login credentials to potential attackers. The bot can be configured to only respond to authorized users and commands, reducing the risk of unauthorized access.
Flexibility: With a Telegram bot, you can customize the commands and responses to fit your specific needs. This means you can create a bot that is tailored to your workflow and allows you to perform tasks more efficiently.

Overall, using a Telegram bot with access to your Linux server’s terminal can provide numerous benefits in terms of convenience, security, and efficiency. It allows you to manage your server remotely and quickly respond to any issues that arise, improving the overall performance and reliability of your system.

There is my code for bot which do all I need:

import os
import subprocess
from aiogram import Bot, Dispatcher, types
from aiogram.dispatcher.filters import Text, IDFilter
from aiogram.types import ReplyKeyboardMarkup, KeyboardButton
from aiogram.types import ParseMode
from aiogram.utils import executor
from aiogram.dispatcher import FSMContext
#from aiogram.dispatcher.filters.state import State, StatesGroup
from aiogram.dispatcher.filters.state import State, StatesGroup
from aiogram.contrib.fsm_storage.memory import MemoryStorage



# Set your Telegram bot token
BOT_TOKEN = '559589BOT=TOCKEN=HEREmf217yg6fg769fg8DMiXCsqs'

# Set the name of the service you want to start or stop
SERVICE_NAME = 'Inception.service'
services = ['Inception.service','gunicorn','nginx','bank','Ai.service','EvgDevBot','exit']
selections = ['YES','NO','exit']
keyboard_mode = types.InlineKeyboardMarkup()
button1 = types.InlineKeyboardButton(text="START", callback_data="start")
button2 = types.InlineKeyboardButton(text="STOP", callback_data="stop")
button3 = types.InlineKeyboardButton(text="RESTART", callback_data="restart")
button4 = types.InlineKeyboardButton(text="STATUS", callback_data="status")
button5 = types.InlineKeyboardButton(text="EXIT", callback_data="exit")
keyboard_mode.add(button1, button2, button3, button4, button5)

keyboard_service = types.InlineKeyboardMarkup()
for service in services:
    keyboard_service.add(types.InlineKeyboardButton(text=service, callback_data=service))

keyboard_confirm = types.InlineKeyboardMarkup()
for selection in selections:
    keyboard_confirm.add(types.InlineKeyboardButton(text=selection, callback_data=selection))

button11 = KeyboardButton('/start')
button21 = KeyboardButton('/service')
button31 = KeyboardButton('/startij')
button41 = KeyboardButton('/stopij')

#keyboard buttons for every selected action

keyboard_main = ReplyKeyboardMarkup(resize_keyboard=True, one_time_keyboard=True).add(button11).add(button21).add(button31).add(button41)


bot = Bot(token=BOT_TOKEN)
dp = Dispatcher(bot)

allowed_ids = [244354634,363636363]

class MyIDFilter(IDFilter):
    pass

id_filter = IDFilter(allowed_ids)

# Create a dispatcher object
storage = MemoryStorage()
dp = Dispatcher(bot, storage=storage)

# Define the states for the conversation
class Action(StatesGroup):
    choose_mode = State()
    choose_service = State()
    confirm = State()

# Define a handler for start
@dp.message_handler(id_filter, commands=['service'])
async def start_message(message: types.Message):
    await message.reply("Choose action", reply_markup=keyboard_mode)
    await bot.delete_message(chat_id=message.chat.id, message_id=message.message_id)
    await bot.delete_message(chat_id=message.chat.id, message_id=message.message_id-1)
    await Action.choose_mode.set()

# Define a handler for choose mode
@dp.callback_query_handler(id_filter, state=Action.choose_mode)
async def selecting_mode(callback_query: types.CallbackQuery, state: FSMContext):
#    await bot.delete_message(chat_id=callback_query.message.chat.id, message_id=callback_query.message.message_id-1)
    await bot.delete_message(chat_id=callback_query.message.chat.id, message_id=callback_query.message.message_id)
    if callback_query.data == 'exit':
        await dp.current_state().reset_state()
        await dp.current_state().set_state(None)
        await callback_query.message.answer("Main menu",reply_markup=keyboard_main)
        await bot.delete_message(chat_id=callback_query.message.chat.id, message_id=callback_query.message.message_id)
    else:
        mode = callback_query.data
        await state.update_data(mode=mode)
        await callback_query.message.answer("Choose service", reply_markup=keyboard_service)
        await Action.choose_service.set()

# Define a handler for choose service
@dp.callback_query_handler(id_filter, state=Action.choose_service)
async def selecting_service(callback_query: types.CallbackQuery, state: FSMContext):
    await bot.delete_message(chat_id=callback_query.message.chat.id, message_id=callback_query.message.message_id)
#    await bot.delete_message(chat_id=callback_query.message.chat.id, message_id=callback_query.message.message_id-1)
    if callback_query.data == 'exit':
        await dp.current_state().reset_state()
        await dp.current_state().set_state(None)
        await callback_query.message.answer("Main menu",reply_markup=keyboard_main)
        await bot.delete_message(chat_id=callback_query.message.chat.id, message_id=callback_query.message.message_id)
    else:
        service = callback_query.data
        async with state.proxy() as data:
            mode = data['mode']
            data['service'] = service
        await callback_query.message.answer(f"Do you want {mode} {service} ?", reply_markup=keyboard_confirm)
        await Action.confirm.set()

# Define confirmed action
@dp.callback_query_handler(id_filter, state=Action.confirm)
async def confirmed(callback_query: types.CallbackQuery, state: FSMContext):
    await bot.delete_message(chat_id=callback_query.message.chat.id, message_id=callback_query.message.message_id)
#    await bot.delete_message(chat_id=callback_query.message.chat.id, message_id=callback_query.message.message_id-1)
    if callback_query.data == 'exit':
        await bot.delete_message(chat_id=callback_query.message.chat.id, message_id=callback_query.message.message_id)
        await dp.current_state().reset_state()
        await dp.current_state().set_state(None)
        await callback_query.message.answer("Main menu:",reply_markup=keyboard_main)
    else:
        data = await state.get_data()
        if callback_query.data == 'YES':
            if data['mode'] == 'status':
                try:
                    output = subprocess.run(['sudo', 'systemctl', data['mode'], data['service']], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
                    await callback_query.message.answer(f"Command output:\n\n{output.stdout.decode()}", reply_markup=keyboard_mode)
                    await Action.choose_mode.set()
                except:
                    await callback_query.message.answer(f"Error", reply_markup=keyboard_mode)
                    await Action.choose_mode.set()

            else:
                subprocess.run(['sudo', 'systemctl', data['mode'], data['service']])
                await callback_query.message.answer(f"{data['mode']} {data['service']} OK! Please choose an action:", reply_markup=keyboard_mode)
                await Action.choose_mode.set()
        else:
            await callback_query.message.answer(f"{data['mode']} {data['service']} CANCELLED! Please choose an action:", reply_markup=keyboard_mode)
            await Action.choose_mode.set()    

# Define a command to start the service
@dp.message_handler(id_filter, commands=['startij'])
async def start_service(message: types.Message):
    subprocess.run(['sudo', 'systemctl', 'start', SERVICE_NAME])
    await message.reply(f'Started {SERVICE_NAME}', reply_markup=keyboard_main)
    await bot.delete_message(chat_id=message.chat.id, message_id=message.message_id)
    await bot.delete_message(chat_id=message.chat.id, message_id=message.message_id-1)
# Define a command to stop the service
@dp.message_handler(id_filter, commands=['stopij'])
async def stop_service(message: types.Message):
    subprocess.run(['sudo', 'systemctl', 'stop', SERVICE_NAME])
    await message.reply(f'Stopped {SERVICE_NAME}', reply_markup=keyboard_main)
    await bot.delete_message(chat_id=message.chat.id, message_id=message.message_id)
    await bot.delete_message(chat_id=message.chat.id, message_id=message.message_id-1)
# Define a command to display the bot's help message
@dp.message_handler(id_filter, commands=['help'])
async def help(message: types.Message):
    help_text = "This bot can start and stop an Ubuntu service.\n\n" \
                "Available commands:\n" \
                "/service - Work with services\n" \
                "/startij - Start the specified service\n" \
                "/stopij - Stop the specified service\n" \
                "/help - Display this help message"
    await message.reply(help_text, parse_mode=ParseMode.MARKDOWN, reply_markup=keyboard_main)
    await bot.delete_message(chat_id=message.chat.id, message_id=message.message_id)
    await bot.delete_message(chat_id=message.chat.id, message_id=message.message_id-1)
# Define a handler for invalid commands
@dp.message_handler(id_filter)
async def invalid_command(message: types.Message):
    await message.reply("Invalid command. Type /help for a list of available commands.", reply_markup=keyboard_main)
    await bot.delete_message(chat_id=message.chat.id, message_id=message.message_id)
    await bot.delete_message(chat_id=message.chat.id, message_id=message.message_id-1)
if __name__ == '__main__':
    executor.start_polling(dp, skip_updates=True)

This is a Python program that allows users to interact with a Telegram bot to manage services on a Linux machine. The program uses the aiogram library to interact with the Telegram API and the subprocess library to execute shell commands on the Linux machine.
The program starts by defining some constants such as the Telegram bot token and the name of the service to manage. It then creates some keyboard layouts using the aiogram library to provide a menu of options for the user.
Next, the program creates an instance of the aiogram Bot class using the bot token and a Dispatcher class to handle incoming messages and callbacks from the user.
The program defines some filters to restrict access to the bot and creates a conversation using the aiogram State class. The conversation is broken down into several steps, where the user selects the action to perform, the service to manage, and confirms the action.
The program defines message handlers for each step of the conversation and uses the aiogram library to send messages to the user and receive input from the user.
When the user selects an action to perform, the program uses the subprocess library to execute the corresponding shell command on the Linux machine. The program then sends a message to the user with the output of the command.
Finally, the program uses the aiogram library to listen for incoming messages from the user and starts the conversation when the user sends the /service command.

Server control by Telegram Bot – Run Shell commands by Python

Leave a Reply Cancel reply

My stats:

Comments