⚠️ Sicherheitsrisiko: Ein Bot, der Shell-Befehle ausführen darf, ist quasi ein offenes Terminal. Wenn jemand dein Token herausfindet oder deinen Bot auf Telegram anschreibt, hat er die volle Kontrolle. Daher muss im Code eine strikte User-ID-Sperre (Whitelist) eingebaut sein, damit der Bot ausschließlich auf deine Nachrichten reagiert.

Hier ist ein kompakter Leitfaden, wie du dir deinen eigenen Admin-Bot mit python-telegram-bot und uv aufbaust.

1. Bot erstellen & Telegram-ID herausfinden

  1. Schreibe den @BotFather auf Telegram an und erstelle mit /newbot einen neuen Bot. Kopiere das API Token.

  2. Schreibe den Bot @userinfobot an, um deine persönliche Telegram User ID (eine lange Zahl) zu erfahren. Nur diese ID darf den Bot später steuern.

2. Projekt mit uv aufsetzen

Da du uv nutzt, erstellen wir eine isolierte Umgebung im Verzeichnis deiner Wahl (z.B. /opt/tg_vps_bot oder in deinem Home-Verzeichnis):

BASH
# Projektordner erstellen und betreten
mkdir -p ~/tg_vps_bot && cd ~/tg_vps_bot

# Virtuelle Umgebung mit uv initialisieren und lib installieren
uv venv
uv pip install python-telegram-bot

Erstelle nun mit deinem Standard-Editor micro die Konfigurationsdatei .env:

BASH
micro .env

Füge deine Daten ein:

TXT
BOT_TOKEN="DEIN_TELEGRAM_BOT_TOKEN"
ALLOWED_USER_ID=123456789  # Deine echte Telegram-ID (als Zahl)

3. Der Bot-Code (bot.py)

Erstelle die Datei bot.py (auf Englisch, wie gewünscht):

BASH
micro bot.py

Hier ist ein asynchrones Python-Skript, das Updates prüft, Systemd-Services neustarten kann und die letzten Zeilen von Logs ausgibt.

PYTHON
import os
import subprocess
from functools import wraps
from telegram import Update
from telegram.ext import ApplicationBuilder, CommandHandler, ContextTypes

# Load environment variables manually to avoid extra dependencies
BOT_TOKEN = os.getenv("BOT_TOKEN")
ALLOWED_USER_ID = int(os.getenv("ALLOWED_USER_ID", 0))

def restricted(func):
    """Decorator to allow execution only for the specified ALLOWED_USER_ID."""

    @wraps(func)
    async def wrapped(
        update: Update, context: ContextTypes.DEFAULT_TYPE, *args, **kwargs
    ):
        user_id = update.effective_user.id
        if user_id != ALLOWED_USER_ID:
            print(f"Unauthorized access attempt by user_id: {user_id}")
            await update.message.reply_text("❌ Unauthorized access.")
            return
        return await func(update, context, *args, **kwargs)

    return wrapped

def run_shell_command(command: list[str]) -> str:
    """Helper to safely run shell commands and return stdout or stderr."""
    try:
        result = subprocess.run(
            command, capture_output=True, text=True, check=False, timeout=30
        )
        output = result.stdout if result.stdout else result.stderr
        return output if output else "Command executed with no output."
    except subprocess.TimeoutExpired:
        return "❌ Error: Command timed out after 30 seconds."
    except Exception as e:
        return f"❌ Python Error: {str(e)}"

@restricted
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Send a welcome message with available commands."""
    help_text = (
        "🤖 **VPS Admin Bot Active**\n\n"
        "Available commands:\n"
        "/check_updates - Check for pending Debian packages\n"
        "/restart_service <name> - Restart a systemd service\n"
        "/view_log <nginx|syslog> - Get the last 20 log lines"
    )
    await update.message.reply_text(help_text, parse_mode="Markdown")

@restricted
async def check_updates(
    update: Update, context: ContextTypes.DEFAULT_TYPE
) -> None:
    """Check for security and package updates."""
    await update.message.reply_text("🔄 Fetching updates... please wait.")
    # Run apt update followed by apt list --upgradable
    subprocess.run(["sudo", "apt-get", "update"], capture_output=True)
    output = run_shell_command(["apt", "list", "--upgradable"])

    # Truncate if output is too long for Telegram (max 4096 chars)
    formatted_output = f"```\n{output[:3500]}\n```"
    await update.message.reply_text(formatted_output, parse_mode="MarkdownV2")

@restricted
async def restart_service(
    update: Update, context: ContextTypes.DEFAULT_TYPE
) -> None:
    """Restart a systemd service (e.g., nginx, php-fpm)."""
    if not context.args:
        await update.message.reply_text(
            "❌ Please specify a service. Example: `/restart_service nginx`"
        )
        return

    service_name = context.args[0]
    await update.message.reply_text(f"⚙️ Restarting {service_name}...")

    cmd = ["sudo", "systemctl", "restart", service_name]
    output = run_shell_command(cmd)

    if "Error" in output or "failed" in output:
        await update.message.reply_text(f"❌ Failed:\n```\n{output}\n```")
    else:
        await update.message.reply_text(f"✅ Service `{service_name}` restarted.")

@restricted
async def view_log(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Read the tail of a specified log file."""
    if not context.args:
        await update.message.reply_text(
            "❌ Specify log type: `/view_log nginx` or `/view_log syslog`"
        )
        return

    log_type = context.args[0].lower()

    if log_type == "nginx":
        # Adjust path matching your ISPConfig/Nginx default setup
        log_path = "/var/log/nginx/error.log"
    elif log_type == "syslog":
        log_path = "/var/log/syslog"
    else:
        await update.message.reply_text("❌ Unknown log type. Use nginx or syslog.")
        return

    cmd = ["sudo", "tail", "-n", "20", log_path]
    output = run_shell_command(cmd)

    await update.message.reply_text(
        f"📋 Last 20 lines of {log_type}:\n```\n{output[:3900]}\n```",
        parse_mode="MarkdownV2",
    )

if __name__ == "__main__":
    if not BOT_TOKEN or ALLOWED_USER_ID == 0:
        print("Error: Please set BOT_TOKEN and ALLOWED_USER_ID in .env")
        exit(1)

    app = ApplicationBuilder().token(BOT_TOKEN).build()

    app.add_handler(CommandHandler("start", start))
    app.add_handler(CommandHandler("check_updates", check_updates))
    app.add_handler(CommandHandler("restart_service", restart_service))
    app.add_handler(CommandHandler("view_log", view_log))

    print("Bot is running...")
    app.run_polling()

4. Rechte & Sudo-Konfiguration

Da der Bot Befehle wie apt-get oder systemctl restart ausführen muss, benötigt der Linux-User, unter dem der Bot läuft, passwortlose sudo-Rechte für genau diese Befehle.

Wenn du den Bot unter deinem normalen User (z.B. deinname) ausführst, öffne die Sudoers-Datei:

BASH
sudo visudo

Füge am Ende folgende Zeile hinzu (ersetze deinname mit deinem Linux-Username):

PLAINTEXT
deinname ALL=(ALL) NOPASSWD: /usr/bin/apt-get update, /usr/bin/apt list --upgradable, /usr/bin/systemctl restart *, /usr/bin/tail -n 20 *

Hinweis: Das erlaubt dem User nur diese spezifischen administrativen Aufgaben ohne Passworteingabe.

5. Bot testen und dauerhaft laufen lassen

Zum Testen kannst du die Umgebungsvariablen laden und den Bot manuell starten:

BASH
export $(cat .env | xargs)
uv run bot.py

Wenn alles klappt, erstelle am besten einen Systemd Service, damit der Bot im Hintergrund läuft und nach einem Server-Reboot automatisch startet:

BASH
sudo micro /etc/systemd/system/tgbot.service

Inhalt (Pfade anpassen):

TXT
[Unit]
Description=Telegram VPS Admin Bot
After=network.target

[Service]
Type=simple
User=deinname
WorkingDirectory=/home/deinname/tg_vps_bot
EnvironmentFile=/home/deinname/tg_vps_bot/.env
ExecStart=/home/deinname/tg_vps_bot/.venv/bin/python bot.py
Restart=always

[Install]
WantedBy=multi-user.target

Service aktivieren und starten:

BASH
sudo systemctl daemon-reload
sudo systemctl enable tgbot.service
sudo systemctl start tgbot.service