"""Persistent storage for raw incoming Telegram updates (development diagnostics).""" from __future__ import annotations import asyncio from contextlib import closing import json import logging import os import re import sqlite3 from pathlib import Path from typing import Any, Awaitable, Callable from aiogram import BaseMiddleware from aiogram.types import TelegramObject, Update logger = logging.getLogger(__name__) RAW_UPDATES_DB_PATH = os.getenv("RAW_UPDATES_DB_PATH", "/app/data/bot_updates.sqlite3") RAW_UPDATE_RETENTION_DAYS = 30 _REDACTED = "[REDACTED]" _SECRET_KEYS = { "authorization", "api_key", "access_token", "bot_token", "token", "secret", "password", } _SECRET_SUFFIXES = ("_api_key", "_access_token", "_bot_token", "_secret", "_password") def _sanitize(value: Any) -> Any: if isinstance(value, dict): result = {} for key, item in value.items(): normalized = str(key).casefold() if normalized in _SECRET_KEYS or normalized.endswith(_SECRET_SUFFIXES): result[key] = _REDACTED else: result[key] = _sanitize(item) return result if isinstance(value, list): return [_sanitize(item) for item in value] if isinstance(value, str): value = re.sub(r"(?i)bearer\s+[^\s\"']+", "Bearer [REDACTED]", value) value = re.sub(r"(?i)(api[_-]?key|access[_-]?token|bot[_-]?token|secret|password)(\s*[=:]\s*)[^\s,;\"']+", r"\1\2[REDACTED]", value) value = re.sub(r"(?i)(?:sk-[A-Za-z0-9_-]{12,}|[0-9]{8,}:[A-Za-z0-9_-]{20,})", _REDACTED, value) return value def _connect(db_path: str | None = None) -> sqlite3.Connection: path = Path(db_path or os.getenv("RAW_UPDATES_DB_PATH", RAW_UPDATES_DB_PATH)) path.parent.mkdir(parents=True, exist_ok=True) connection = sqlite3.connect(path, timeout=10) connection.row_factory = sqlite3.Row connection.execute("PRAGMA busy_timeout = 10000") return connection def init_raw_updates_db(db_path: str | None = None) -> None: with closing(_connect(db_path)) as connection, connection: connection.execute( """CREATE TABLE IF NOT EXISTS raw_updates ( update_id INTEGER PRIMARY KEY, event_type TEXT NOT NULL, payload TEXT NOT NULL, received_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP )""" ) connection.execute( "CREATE INDEX IF NOT EXISTS idx_raw_updates_received_at ON raw_updates(received_at)" ) connection.execute( "DELETE FROM raw_updates WHERE received_at < datetime('now', ?)", (f"-{RAW_UPDATE_RETENTION_DAYS} days",), ) def save_raw_update(update_id: int, payload: dict[str, Any], db_path: str | None = None) -> None: safe_payload = _sanitize(payload) event_type = next( (key for key, value in safe_payload.items() if key != "update_id" and value is not None), "unknown", ) serialized = json.dumps(safe_payload, ensure_ascii=False, separators=(",", ":")) with closing(_connect(db_path)) as connection, connection: connection.execute( "INSERT OR REPLACE INTO raw_updates (update_id, event_type, payload) VALUES (?, ?, ?)", (int(update_id), event_type, serialized), ) connection.execute( "DELETE FROM raw_updates WHERE received_at < datetime('now', ?)", (f"-{RAW_UPDATE_RETENTION_DAYS} days",), ) def get_raw_update(update_id: int, db_path: str | None = None) -> dict[str, Any] | None: with closing(_connect(db_path)) as connection: row = connection.execute( "SELECT update_id, event_type, payload, received_at FROM raw_updates WHERE update_id = ?", (int(update_id),), ).fetchone() if row is None: return None return { "update_id": row["update_id"], "event_type": row["event_type"], "payload": json.loads(row["payload"]), "received_at": row["received_at"], } class RawUpdateMiddleware(BaseMiddleware): async def __call__( self, handler: Callable[[TelegramObject, dict[str, Any]], Awaitable[Any]], event: TelegramObject, data: dict[str, Any], ) -> Any: if isinstance(event, Update): try: await asyncio.to_thread( save_raw_update, event.update_id, event.model_dump(mode="json"), ) except Exception as exc: logger.warning("Raw update persistence failed (%s)", type(exc).__name__) return await handler(event, data)