Reports

# code/python (save as bot.py)

"""

Secure minimal Telegram bot using python-telegram-bot (v20+)

Features:

- Token loaded from env var

- Admin-only commands (by telegram user_id)

- Safe DB access (sqlite + parameterized queries)

- Graceful error handling & rate limiting (simple)

- Example of using webhook (recommended) or polling fallback

"""

import os

import logging

import sqlite3

import time

from functools import wraps

from http import HTTPStatus

from telegram import Update, Bot

from telegram.ext import ApplicationBuilder, CommandHandler, ContextTypes, MessageHandler, filters

# --- Configuration (from environment) ---

BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN")

if not BOT_TOKEN:

raise SystemExit("TELEGRAM_BOT_TOKEN env var required")

ALLOWED_ADMINS = {int(x) for x in os.getenv("BOT_ADMINS", "").split(",") if x.strip()} # comma-separated IDs

WEBHOOK_URL = os.getenv("WEBHOOK_URL") # e.g. https://your.domain/path

DATABASE_PATH = os.getenv("BOT_DB_PATH", "bot_data.sqlite")

# --- Logging ---

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")

logger = logging.getLogger("secure_bot")

# --- DB helpers (safe parameterized queries) ---

def init_db():

conn = sqlite3.connect(DATABASE_PATH, check_same_thread=False)

c = conn.cursor()

c.execute("""CREATE TABLE IF NOT EXISTS messages (

              id INTEGER PRIMARY KEY AUTOINCREMENT,

              user_id INTEGER,

              username TEXT,

              text TEXT,

              ts INTEGER

            )""")

conn.commit()

return conn

db = init_db()

# --- admin check decorator ---

def admin_only(func):

@wraps(func)

async def wrapper(update: Update, context: ContextTypes.DEFAULT_TYPE):

    user = update.effective_user

    if not user or user.id not in ALLOWED_ADMINS:

        logger.warning("Unauthorized access attempt by %s (%s)", user.id if user else "unknown", user.username if user else "")

        if update.effective_chat:

            await update.effective_chat.send_message("Unauthorized.")

        return

    return await func(update, context)

return wrapper

# --- Handlers ---

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):

await update.message.reply_text("Hello. This bot is configured securely. Use /help for commands.")

async def help_cmd(update: Update, context: ContextTypes.DEFAULT_TYPE):

await update.message.reply_text("/status - admin only\\n/echo \<text\> - echo back\\n/help - this message")

@admin_only

async def status(update: Update, context: ContextTypes.DEFAULT_TYPE):

await update.message.reply_text("OK — bot is running.")

async def echo(update: Update, context: ContextTypes.DEFAULT_TYPE):

\# store incoming message safely

try:

    msg = update.effective_message

    with db:

        db.execute("INSERT INTO messages (user_id, username, text, ts) VALUES (?, ?, ?, ?)",

                   (msg.from_user.id, msg.from_user.username or "", msg.text or "", int(time.time())))

    \# simple rate-limit: disallow messages \> 400 chars

    if msg.text and len(msg.text) \> 400:

        await msg.reply_text("Message too long.")

        return

    await msg.reply_text(msg.text or "Empty message.")

except Exception as e:

    logger.exception("Error in echo handler: %s", e)

    await update.effective_chat.send_message("Internal error.")

# --- basic command to rotate token reminder (admin only) ---

@admin_only

async def rotate_reminder(update: Update, context: ContextTypes.DEFAULT_TYPE):

await update.message.reply_text("Reminder: rotate token and update TELEGRAM_BOT_TOKEN env var on server.")

# --- Build application ---

async def main():

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

app.add_handler(CommandHandler("start", start))

app.add_handler(CommandHandler("help", help_cmd))

app.add_handler(CommandHandler("status", status))

app.add_handler(CommandHandler("rotate", rotate_reminder))

app.add_handler(CommandHandler("echo", echo))

app.add_handler(MessageHandler(filters.TEXT & \~filters.COMMAND, echo))

\# Webhook preferred (more secure than polling) if WEBHOOK_URL provided

if WEBHOOK_URL:

    \# set webhook (TLS must be handled by your web server/reverse-proxy)

    bot = Bot(token=BOT_TOKEN)

    await bot.set_webhook(WEBHOOK_URL)

    logger.info("Webhook set to %s", WEBHOOK_URL)

    \# start the app (it will use long polling by default in local runner;

    \# for production you should run an ASGI server with endpoints calling app.update_queue)

    await app.initialize()

    await app.start()

    logger.info("Bot started with webhook mode (app running).")

    \# keep running until terminated

    await app.updater.stop()  # placeholder to keep structure consistent

else:

    \# fallback to polling (useful for dev only)

    logger.info("Starting in polling mode (development only).")

    await app.run_polling()

if _name_ == "_main_":

import asyncio

asyncio.run(main())

79785768