| 1 | from helpers.api import ApiHandler, Request, Response |
| 2 | from helpers.print_style import PrintStyle |
| 3 | from plugins._telegram_integration.helpers.dependencies import ensure_dependencies |
| 4 | |
| 5 | |
| 6 | class TelegramWebhook(ApiHandler): |
| 7 | """Receives Telegram webhook updates. No auth/CSRF — Telegram cannot send session cookies.""" |
| 8 | |
| 9 | @classmethod |
| 10 | def requires_auth(cls) -> bool: |
| 11 | return False |
| 12 | |
| 13 | @classmethod |
| 14 | def requires_csrf(cls) -> bool: |
| 15 | return False |
| 16 | |
| 17 | @classmethod |
| 18 | def get_methods(cls) -> list[str]: |
| 19 | return ["POST"] |
| 20 | |
| 21 | async def process(self, input: dict, request: Request) -> dict | Response: |
| 22 | ensure_dependencies() |
| 23 | from aiogram.types import Update |
| 24 | |
| 25 | from plugins._telegram_integration.helpers.bot_manager import get_bot |
| 26 | |
| 27 | # Identify which bot this update is for |
| 28 | bot_name = request.args.get("bot", "") |
| 29 | if not bot_name: |
| 30 | return Response("Missing ?bot= parameter", 400) |
| 31 | |
| 32 | instance = get_bot(bot_name) |
| 33 | if not instance: |
| 34 | return Response(f"Bot not found: {bot_name}", 404) |
| 35 | |
| 36 | # Verify webhook secret if configured |
| 37 | secret_header = request.headers.get("X-Telegram-Bot-Api-Secret-Token", "") |
| 38 | if instance.webhook_secret and secret_header != instance.webhook_secret: |
| 39 | return Response("Invalid secret token", 403) |
| 40 | |
| 41 | # Parse and feed the update to aiogram |
| 42 | try: |
| 43 | update = Update.model_validate(input, context={"bot": instance.bot}) |
| 44 | await instance.dispatcher.feed_update(instance.bot, update) |
| 45 | except Exception as e: |
| 46 | PrintStyle.error(f"Telegram webhook ({bot_name}): {e}") |
| 47 | return Response("Internal error", 500) |
| 48 | |
| 49 | return {"ok": True} |