Add ability to use self-signed certificate, refactor config.py

mybot/__init__.py

@@ -8,7 +8,6 @@ from .states import get_state_storage
 from .handlers import register_handlers
 from .middlewares import setup_middlewares
 from .filters import add_custom_filters
-from .markup import setup_markup
 from .webhook import create_app
 
 
@@ -22,14 +21,15 @@ def create_bot(config: Config, i18n: I18N, engine):
                   state_storage=state_storage,
                   threaded=False if config.use_webhook else True)
     register_handlers(bot)
-    markup = setup_markup(bot, i18n)
-    setup_middlewares(bot, i18n, engine, markup)
+    setup_middlewares(bot, i18n, engine)
     add_custom_filters(bot, config)
     bot.delete_webhook()
     if config.use_webhook:
         bot.set_webhook(config.webhook.url,
                         drop_pending_updates=config.webhook.drop_pending_updates,
-                        max_connections=config.webhook.max_connections)
+                        max_connections=config.webhook.max_connections,
+                        secret_token=config.webhook.secret_token,
+                        certificate=config.webhook.cert_path)
     return bot
 
 

mybot/config.py

@@ -1,5 +1,7 @@
 import os
+import secrets
 from dataclasses import dataclass
+from typing import Optional
 
 
 @dataclass
@@ -23,21 +25,35 @@ class BotConfig:
 
 @dataclass
 class WebhookConfig:
-    domain: str
+    domain: Optional[str]
     url_path: str
     max_connections: int
     drop_pending_updates: bool
 
+    # secret token
+    use_secret_token: bool
+    secret_token: Optional[str]
+
+    # self-signed certificate
+    cert_path: Optional[str]
+
+    def __post_init__(self):
+        if self.use_secret_token and not self.secret_token:
+            self.secret_token = secrets.token_hex()
+
     @property
     def url(self):
-        return f"https://{self.domain}/{self.url_path}"
+        return f"https://{self.domain}{self.url_path}"
 
     @classmethod
     def from_env(cls):
         return cls(os.getenv("WEBHOOK_DOMAIN"),
-                   os.getenv("WEBHOOK_URL_PATH"),
+                   os.getenv("WEBHOOK_URL_PATH", "/"),
                    int(os.getenv("WEBHOOK_MAX_CONNECTIONS", 40)),
-                   bool(int(os.getenv("WEBHOOK_DROP_PENDING", True))))
+                   bool(int(os.getenv("WEBHOOK_DROP_PENDING", True))),
+                   bool(int(os.getenv("WEBHOOK_USE_SECRET_TOKEN", True))),
+                   os.getenv("WEBHOOK_SECRET_TOKEN"),
+                   os.getenv("WEBHOOK_CERT_PATH"))
 
 
 @dataclass

mybot/handlers/basic.py

@@ -1,13 +1,9 @@
 from telebot import TeleBot
-from telebot.types import Message, CallbackQuery
+from telebot.types import Message
 
 
-def start(message: Message, bot: TeleBot, t, m, **kwargs):
-    bot.send_message(message.chat.id, t("start"), reply_markup=m("start"))
-
-
-def start_call(call: CallbackQuery, bot: TeleBot, t, m, **kwargs):
-    bot.send_message(call.message.chat.id, t("start"), reply_markup=m("start"))
+def start(message: Message, bot: TeleBot, t, **kwargs):
+    bot.send_message(message.chat.id, t("start"))
 
 
 def help_(message, bot, t, **kwargs):
@@ -17,5 +13,3 @@ def help_(message, bot, t, **kwargs):
 def register_handlers(bot: TeleBot):
     bot.register_message_handler(start, commands=["start"], pass_bot=True)
     bot.register_message_handler(help_, commands=["help"], pass_bot=True)
-
-    bot.register_callback_query_handler(start_call, lambda call: call.data == "start")

mybot/markup/__init__.py

@@ -1,11 +0,0 @@
-from telebot import TeleBot
-
-from ..i18n import I18N
-from .base import MarkupManager
-from .simple import SimpleMarkup
-
-
-def setup_markup(bot: TeleBot, i18n: I18N):
-    markup_mg = MarkupManager(bot, i18n)
-    markup_mg.register_prototype(SimpleMarkup)
-    return markup_mg

mybot/markup/base.py

@@ -1,55 +0,0 @@
-import abc
-from typing import Optional, Type
-
-from telebot import TeleBot
-from telebot.types import CallbackQuery, InlineKeyboardMarkup
-
-from ..i18n import I18N
-
-
-class Markup (metaclass=abc.ABCMeta):
-    tag: str
-
-    def __init__(self, bot: TeleBot, i18n: I18N):
-        self.bot = bot
-        self.t = i18n
-
-    def __call__(self, *args, **kwargs):
-        return self.build(*args, **kwargs)
-
-    def check(self, call: CallbackQuery) -> bool:
-        return call.data == self.tag
-
-    @abc.abstractmethod
-    def build(self, *args, **kwargs) -> Optional[InlineKeyboardMarkup]:
-        pass
-
-
-class DummyMarkup (Markup):
-    tag = "__dummy"
-
-    def check(self, call: CallbackQuery) -> bool:
-        return True
-
-    def build(self, *args, **kwargs) -> Optional[InlineKeyboardMarkup]:
-        return None
-
-
-class MarkupManager:
-    def __init__(self, bot: TeleBot, i18n: I18N):
-        self.bot = bot
-        self.i18n = i18n
-        self._prototypes: list[Markup] = []
-        self._dummy = DummyMarkup(self.bot, self.i18n)
-
-    def register_prototype(self, markup_proto_class: Type[Markup]):
-        self._prototypes.append(markup_proto_class(self.bot, self.i18n))
-
-    def __call__(self, tag: str, *args, **kwargs):
-        return self.get(tag).build(*args, **kwargs)
-
-    def get(self, tag: str) -> Optional[Markup]:
-        for mp in self._prototypes:
-            if mp.tag == tag:
-                return mp
-        return None

mybot/markup/simple.py

@@ -1,15 +0,0 @@
-from typing import Optional
-
-from telebot.types import InlineKeyboardMarkup, InlineKeyboardButton
-
-from .base import Markup
-
-
-class SimpleMarkup (Markup):
-    tag = "start"
-
-    def build(self, *args, **kwargs) -> Optional[InlineKeyboardMarkup]:
-        return (InlineKeyboardMarkup()
-                .add(InlineKeyboardButton("start", callback_data="start"))
-                .add(InlineKeyboardButton("help", callback_data="help"))
-                )

mybot/middlewares/__init__.py

@@ -5,6 +5,6 @@ from .arguments import ArgumentsMiddleware
 from .database import DatabaseMiddleware
 
 
-def setup_middlewares(bot: TeleBot, i18n: I18N, engine, markup):
-    bot.setup_middleware(ArgumentsMiddleware(i18n, markup))
+def setup_middlewares(bot: TeleBot, i18n: I18N, engine):
+    bot.setup_middleware(ArgumentsMiddleware(i18n))
     bot.setup_middleware(DatabaseMiddleware(engine))

mybot/middlewares/arguments.py

@@ -3,24 +3,16 @@ from telebot.types import Message, CallbackQuery
 
 
 class ArgumentsMiddleware (BaseMiddleware):
-    def __init__(self, i18n, markup):
+    def __init__(self, i18n):
         super().__init__()
         self.i18n = i18n
-        self.markup = markup
         self.update_types = ["message", "callback_query"]
 
     def pre_process(self, obj, data: dict):
         if isinstance(obj, Message):
-            self.pre_process_message(obj, data)
+            data["t"] = self.i18n.customized_call(message=obj)
         elif isinstance(obj, CallbackQuery):
-            self.pre_process_callback(obj, data)
-        data["m"] = self.markup
-
-    def pre_process_message(self, message: Message, data: dict):
-        data["t"] = self.i18n.customized_call(message=message)
-
-    def pre_process_callback(self, call: CallbackQuery, data: dict):
-        data["t"] = self.i18n.customized_call(callback=call)
+            data["t"] = self.i18n.customized_call(callback=obj)
 
     def post_process(self, message, data: dict, exception: BaseException):
         pass

mybot/webhook/__init__.py

@@ -1,35 +1,35 @@
-from flask import Flask, Blueprint, request, abort, g
+from flask import Flask, request, abort, g
 from telebot import TeleBot
 from telebot.types import Update
 
 from ..config import Config
 
 
-bot_bp = Blueprint("bot", __name__)
-
-
-@bot_bp.route("/", methods=["GET", "POST"])
 def handle_updates():
     if request.method == "GET":
-        abort(404)
+        abort(404)  # safer to 404
+    if g.config.webhook.use_secret_token:
+        if request.headers.get("X-Telegram-Bot-Api-Secret-Token") != g.config.webhook.secret_token:
+            abort(404)
     if request.headers.get("content-type") == "application/json":
         update = Update.de_json(request.get_json())
         g.bot.process_new_updates([update])
         return ""
     else:
-        abort(403)
+        abort(404)  # safer to 404
 
 
-def inject_g(bot: TeleBot, config: Config):
+def inject_g(**kwargs):
     def inner():
-        g.bot = bot
-        g.config = config
+        for k, v in kwargs.items():
+            setattr(g, k, v)
     return inner
 
 
 def create_app(bot: TeleBot, config: Config):
     app = Flask(__name__)
-    app.register_blueprint(bot_bp, url_prefix=f"{config.webhook.url_path}")
-    app.before_request(inject_g(bot, config))
-
+    app.add_url_rule(config.webhook.url_path,
+                     view_func=handle_updates,
+                     methods=["GET", "POST"])
+    app.before_request(inject_g(bot=bot, config=config))
     return app