# 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())
