#!/usr/bin/env python3 """ Webhook-Queue-Integration — Signatur-Prüfung + Enqueue. ======================================================= Wird in app.py importiert und wickelt die Webhook-Routen ab. Ersetzt NICHT die bestehende Logik, sondern sitzt DAVOR. Flow: 1. Webhook empfängt Request 2. Signatur wird geprüft (WhatsApp X-Hub-Signature-256) 3. Message-IDs werden extrahiert 4. Submission wird in Queue eingereiht (SQLite UNIQUE = Dedup) 5. 200 OK wird sofort zurückgegeben 6. Worker (queue_worker.py) verarbeitet asynchron Usage in app.py: from webhook_queue import whatsapp_intake, telegram_intake @app.route("/whatsapp/webhook", methods=["GET", "POST"]) def whatsapp_webhook(): ... return whatsapp_intake(request, db) @app.route("/telegram/webhook", methods=["GET", "POST"]) def telegram_webhook(): ... return telegram_intake(request, db) """ import hashlib import hmac import logging import os from typing import Optional logger = logging.getLogger(__name__) # ── Konfiguration ──────────────────────────────────────────────────── META_APP_SECRET = os.environ.get("META_APP_SECRET", "") # Falls kein App Secret gesetzt → Signatur-Prüfung überspringen (Test-Modus) # ── Signatur-Prüfung ───────────────────────────────────────────────── def verify_whatsapp_signature( raw_body: bytes, signature_header: Optional[str], ) -> bool: """ Prüft die X-Hub-Signature-256 von Meta. Args: raw_body: Der rohe Request-Body als Bytes. signature_header: Wert des X-Hub-Signature-256-Headers. Returns: True wenn gültig oder Test-Modus (kein Secret). """ if not META_APP_SECRET: logger.debug("Kein META_APP_SECRET → Signatur-Prüfung übersprungen") return True if not signature_header: logger.warning("X-Hub-Signature-256 Header fehlt") return False if not signature_header.startswith("sha256="): logger.warning("Unbekanntes Signatur-Format: %s...", signature_header[:30]) return False expected_hex = signature_header[7:] # Nach "sha256=" computed_hmac = hmac.new( key=META_APP_SECRET.encode("utf-8"), msg=raw_body, digestmod=hashlib.sha256, ) computed_hex = computed_hmac.hexdigest() if not hmac.compare_digest(computed_hex, expected_hex): logger.warning( "❌ WhatsApp-Signatur UNGÜLTIG — möglicher Replay-Angriff" ) return False logger.debug("✅ WhatsApp-Signatur gültig") return True # ── Intake-Funktionen ──────────────────────────────────────────────── def whatsapp_intake(request_obj, db) -> tuple: """ Verarbeitet EINE WhatsApp-Webhook-Anfrage. Args: request_obj: Flask request-Objekt. db: SQLite-Datenbankverbindung. Returns: (body, status_code) — IMMER 200 OK. """ from queue_handler import QueueHandler qh = QueueHandler(db) # ── 1. Signatur prüfen ────────────────────────────────────────── raw_body = request_obj.get_data() signature = request_obj.headers.get("X-Hub-Signature-256", "") if not verify_whatsapp_signature(raw_body, signature): logger.warning("WhatsApp-Request mit ungültiger Signatur abgelehnt") return "Forbidden", 403 # ── 2. Body parsen ────────────────────────────────────────────── body = request_obj.get_json(force=True, silent=True) if not body: return "OK", 200 entries = body.get("entry", []) enqueued = 0 duplicates = 0 # ── 3. Jede Nachricht in Queue ────────────────────────────────── for entry in entries: changes = entry.get("changes", []) for change in changes: value = change.get("value", {}) messages = value.get("messages", []) for msg in messages: msg_id = msg.get("id", "") if not msg_id: continue # Payload bauen (nur das relevante message-Objekt) payload = { "entry": [{ "id": entry.get("id", ""), "changes": [{ "value": { "messaging_product": value.get("messaging_product", "whatsapp"), "metadata": value.get("metadata", {}), "contacts": value.get("contacts", []), "messages": [msg], } }] }] } success, status = qh.enqueue( channel="whatsapp", message_id=msg_id, payload=payload, raw_body=raw_body, ) if success: enqueued += 1 else: duplicates += 1 logger.info("WhatsApp-Duplikat: %s → %s", msg_id, status) logger.info( "WhatsApp-Intake: %d enqueued, %d duplicates", enqueued, duplicates, ) return "OK", 200 def telegram_intake(request_obj, db) -> tuple: """ Verarbeitet EINE Telegram-Webhook-Anfrage. Args: request_obj: Flask request-Objekt. db: SQLite-Datenbankverbindung. Returns: (body, status_code) — IMMER 200 OK. """ from queue_handler import QueueHandler qh = QueueHandler(db) body = request_obj.get_json(force=True, silent=True) if not body: return "OK", 200 msg = body.get("message", {}) if not msg: # Kein Message-Objekt (z.B. edited_message, callback_query) update_id = str(body.get("update_id", "unknown")) logger.debug("Telegram-Update ohne Message: update_id=%s", update_id) return "OK", 200 msg_id = str(msg.get("message_id", "")) if not msg_id: return "OK", 200 # Payload ist das gesamte Update (enthält update_id + message) success, status = qh.enqueue( channel="telegram", message_id=msg_id, payload=body, raw_body=request_obj.get_data(), ) if success: logger.info("Telegram-Intake: %s enqueued", msg_id) else: logger.info("Telegram-Duplikat: %s → %s", msg_id, status) return "OK", 200 # ── Queue-Status (für Monitoring) ──────────────────────────────────── def get_queue_status(db) -> dict: """Gibt Queue-Statistiken zurück (für /admin/queue Endpoint).""" from queue_handler import QueueHandler qh = QueueHandler(db) return qh.get_queue_stats()