Telegram push

.vscode/launch.json

@@ -0,0 +1,17 @@
+{
+    // Use IntelliSense to learn about possible attributes.
+    // Hover to view descriptions of existing attributes.
+    // For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
+    "version": "0.2.0",
+    "configurations": [
+        {
+            "name": "pi-manage runserver",
+            "type": "python",
+            "request": "launch",
+            "program": "${workspaceFolder}/pi-manage",
+            "console": "integratedTerminal",
+            "args": [ "runserver" ],
+            "justMyCode": false
+        }
+    ]
+}

flake.nix

@@ -0,0 +1,31 @@
+{
+  inputs.nixpkgs.url = "nixpkgs";
+  inputs.makeShell.url = "github:ursi/nix-make-shell";
+
+  outputs = { self, nixpkgs, makeShell }:
+    let
+      supportedSystems = [ "x86_64-linux" ];
+      forAllSystems = nixpkgs.lib.genAttrs supportedSystems;
+      pkgs = forAllSystems (system: nixpkgs.legacyPackages.${system});
+    in
+    {
+      packages = forAllSystems (system: {
+        default = pkgs.${system}.poetry2nix.mkPoetryApplication { projectDir = self; };
+      });
+
+      devShells = forAllSystems (system: 
+      let 
+        make-shell = import makeShell { inherit system; pkgs = pkgs.${system};};
+      in
+      {
+        default = make-shell {
+          packages = with pkgs.${system}; [
+            python310
+            jetbrains.pycharm-community
+            telegram-bot-api 
+            vscode-fhs
+          ];
+        };
+      });
+    };
+}

privacyidea/api/lib/prepolicy.py

@@ -73,6 +73,7 @@
 from flask import g, current_app
 from privacyidea.lib.policy import SCOPE, ACTION, REMOTE_USER
 from privacyidea.lib.policy import Match, check_pin
+from privacyidea.lib.tokens.telegrampushtoken import TELEGRAM_PUSH_ACTION
 from privacyidea.lib.user import (get_user_from_param, get_default_realm,
                                   split_user, User)
 from privacyidea.lib.token import (get_tokens, get_realms_of_token, get_token_type, get_token_owner)
@@ -1464,13 +1465,18 @@ def pushtoken_wait(request, action):
     :return:
     """
     user_object = request.User
-    waiting = Match.user(g, scope=SCOPE.AUTH, action=PUSH_ACTION.WAIT,
+    check_push_wait(request, user_object, PUSH_ACTION.WAIT)
+    check_push_wait(request, user_object, TELEGRAM_PUSH_ACTION.WAIT)
+
+
+def check_push_wait(request, user_object, action):
+    waiting = Match.user(g, scope=SCOPE.AUTH, action=action,
                          user_object=user_object if user_object else None)\
         .action_values(unique=True, allow_white_space_in_action=True)
     if len(waiting) >= 1:
-        request.all_data[PUSH_ACTION.WAIT] = int(list(waiting)[0])
+        request.all_data[action] = int(list(waiting)[0])
     else:
-        request.all_data[PUSH_ACTION.WAIT] = False
+        request.all_data[action] = False
 
 
 def pushtoken_add_config(request, action):

privacyidea/api/lib/utils.py

@@ -25,6 +25,8 @@
 #
 import string
 
+import flask
+
 from ...lib.error import (ParameterError,
                           AuthError, ERROR)
 from ...lib.log import log_with
@@ -272,7 +274,7 @@ def check_unquote(request, data):
         return copy(data)
 
 
-def get_all_params(request):
+def get_all_params(request: flask.Request):
     """
     Retrieve all parameters from a request, no matter if these are GET or POST requests
     or parameters are contained as viewargs like the serial in DELETE /token/<serial>

privacyidea/api/token.py

@@ -293,7 +293,7 @@ def init():
     """
     response_details = {}
     tokenrealms = None
-    param = request.all_data
+    param: dict = request.all_data
 
     # check admin authorization
     # user_tnum = len(getTokens4UserOrSerial(user))
@@ -314,6 +314,7 @@ def init():
         g.audit_object.log({"success": True})
         # The token was created successfully, so we add token specific
         # init details like the google URL to the response
+        param.update({"g": g})
         init_details = tokenobject.get_init_detail(param, user)
         response_details.update(init_details)
 

privacyidea/api/ttype.py

@@ -70,6 +70,7 @@ def before_request():
     # access_route contains the ip adresses of all clients, hops and proxies.
     g.client_ip = get_client_ip(request,
                                 get_from_config(SYSCONF.OVERRIDECLIENT))
+    g.request_headers = request.headers
     g.serial = getParam(request.all_data, "serial") or None
     g.audit_object.log({"success": False,
                         "action_detail": "",

privacyidea/app.py

@@ -45,6 +45,7 @@
 from privacyidea.api.caconnector import caconnector_blueprint
 from privacyidea.api.register import register_blueprint
 from privacyidea.api.auth import jwtauth
+from privacyidea.lib.telegrambot.telegrambot import TelegramBot
 from privacyidea.webui.login import login_blueprint, get_accepted_language
 from privacyidea.webui.certificate import cert_blueprint
 from privacyidea.api.machineresolver import machineresolver_blueprint
@@ -176,6 +177,7 @@ def create_app(config_name="development",
     app.register_blueprint(subscriptions_blueprint, url_prefix='/subscriptions')
     app.register_blueprint(monitoring_blueprint, url_prefix='/monitoring')
     app.register_blueprint(tokengroup_blueprint, url_prefix='/tokengroup')
+    TelegramBot.init_callbacks()
     db.init_app(app)
     migrate = Migrate(app, db)
 

privacyidea/lib/config.py

@@ -743,6 +743,7 @@ def get_token_list():
     module_list.add("privacyidea.lib.tokens.pushtoken")
     module_list.add("privacyidea.lib.tokens.indexedsecrettoken")
     module_list.add("privacyidea.lib.tokens.webauthntoken")
+    module_list.add("privacyidea.lib.tokens.telegrampushtoken")
 
     # Dynamic token modules
     dynamic_token_modules = get_app_config_value("PI_TOKEN_MODULES")

privacyidea/lib/pirequest.py

@@ -0,0 +1,5 @@
+import flask
+
+
+class PiRequestClass(flask.Request):
+    all_data: dict

privacyidea/lib/telegrambot/telegrambot.py

@@ -0,0 +1,121 @@
+import telebot
+from telebot.types import *
+from telebot.util import quick_markup
+from functools import cache
+from privacyidea.lib import _
+import time
+
+from privacyidea.lib.tokens.telegrampushtoken import PushClickEvent, TelegramMessageData, TelegramPushTokenClass
+
+log = logging.getLogger(__name__)
+
+
+class TelegramBot:
+    """
+    This class wraps a singleton TeleBot instance to communicate with the Bot API
+    It tries to have a unidirectional dependency only to a domain class TelegramPushTokenClass by using callbacks 
+    with late binding
+    """
+    @classmethod
+    def init_callbacks(cls):
+        cls._complete_enrollment = TelegramPushTokenClass._complete_enrollment
+        cls._on_push_click = TelegramPushTokenClass._on_push_click
+        TelegramPushTokenClass._bot_factory = cls.get_instance
+
+    @classmethod
+    @cache
+    def get_instance(cls, bot_api_token: str, bot_api_url: str, bot_webhook_endpoint: str):
+        """
+        :rtype: TelegramBot
+        """
+        return TelegramBot(bot_api_token, bot_api_url, bot_webhook_endpoint)
+
+    def __init__(self, bot_api_token, bot_api_url, bot_webhook_endpoint):
+        # Threaded environment doesn't work well with flask
+        self._bot = telebot.TeleBot(bot_api_token, threaded=False)
+        telebot.apihelper.API_URL = bot_api_url
+        self._bot.register_message_handler(self._handle_start, commands=['start'])
+        self._bot.register_message_handler(self._send_welcome)
+        self._bot.register_callback_query_handler(self._handle_callback, lambda x: True)
+        # Remove webhook, it fails sometimes if there is a previous webhook
+        self._bot.remove_webhook()
+        time.sleep(0.1)
+        # Set webhook
+        self._bot.set_webhook(url=bot_webhook_endpoint, drop_pending_updates=True)
+
+    def _handle_callback(self, callback_query: CallbackQuery):
+        """
+        This method handles a click on inline button
+        """
+        try:
+            event = PushClickEvent(callback_query.data)
+            self._on_push_click(event, lambda msg: self._confirm_callback(callback_query, msg))
+        finally:
+            return
+
+    def _handle_start(self, message: Message):
+        """
+        Handler of a start command - special command to a Telegram bot which is required to start conversation
+        """
+        try:
+            split = message.text.split()
+            if len(split) != 2:
+                # If user manually types /start we give him a hint
+                self._bot.reply_to(message,
+                                _("Hi there, I am Telegram 2FA push bot. "
+                                "To complete registration of your account as a 2nd factor you either have to use QR code "
+                                ", or url link with embedded registration token, or manually run command /start [TOKEN]"))
+                return
+
+            enrollment_credential = split[1]
+            chat_id = message.chat.id
+            # normally users will send a start command during the 2nd step of an enrollment through deep link in a QR code
+            self._complete_enrollment(chat_id, enrollment_credential, lambda msg: self.submit_message(chat_id, msg))
+        finally:
+            return
+
+    def _send_welcome(self, message: Message):
+        """
+        Catch-all handler to greet the user if it sends unexpected things to a bot
+        """
+        self._bot.reply_to(message, _("Hi there, I am Telegram 2FA push bot. You'll need to register your Telegram account"
+                                    " before I can send you 2FA notifications. You have to obtain registration link/QR and" 
+                                    " then use this link to start dialog with me again. "))
+
+    def process_new_update(self, update: Update):
+        """
+        Public method to pipe received json from a webhook to a TeleBot engine
+        """
+        self._bot.process_new_updates([update])
+
+    def submit_message(self, chat_id: int, message: TelegramMessageData):
+        """
+        Public method which consumers can use to send arbitrary text message to a private chat of a bot with the user.
+        Message could also contain 64-symbol data, in which case two inline buttons will be created under the message
+        """
+        try:
+            markup = None
+            if message.callback_data is not None:
+                markup = quick_markup({
+                    'Confirm': {'callback_data': f"C_{message.callback_data}"},
+                    'Decline': {'callback_data': f"D_{message.callback_data}"}
+                }, row_width=2)
+            msg = self._bot.send_message(chat_id, text=message.message, reply_markup=markup)
+            if msg is None:
+                log.warning("Error sending message to a chat %s", chat_id)
+                return False
+            return True
+        except Exception as err:
+            log.error(f"Exception sending message to a chat: {err}")
+            return False
+
+    def _confirm_callback(self, callback: CallbackQuery, text: str):
+        self._bot.answer_callback_query(callback.id, text, show_alert=True)
+
+    @cache
+    def get_bot_name(self):
+        """
+        Asking username of a bot from API so that administrator won't need to set in manually in the policies
+        """
+        user = self._bot.get_me()
+        return user.username

privacyidea/lib/token.py

@@ -65,6 +65,8 @@
 import datetime
 import os
 import logging
+from typing import TypeVar
+
 from six import string_types
 
 from sqlalchemy import (and_, func)
@@ -554,7 +556,10 @@ def get_tokens_paginate(tokentype=None, realm=None, assigned=None, user=None,
     return ret
 
 
-def get_one_token(*args, **kwargs):
+TokenClassT = TypeVar('TokenClassT', bound=TokenClass)
+
+
+def get_one_token(*args, **kwargs) -> TokenClassT:
     """
     Fetch exactly one token according to the given filter arguments, which are passed to
     ``get_tokens``. Raise ``ResourceNotFoundError`` if no token was found. Raise

privacyidea/lib/tokenclass.py

@@ -82,6 +82,8 @@
 import traceback
 from datetime import datetime, timedelta
 
+import flask
+
 from .error import (TokenAdminError,
                     ParameterError)
 
@@ -1689,7 +1691,7 @@ def get_as_dict(self):
         return token_dict
 
     @classmethod
-    def api_endpoint(cls, request, g):
+    def api_endpoint(cls, request: flask.Request, g):
         """
         This provides a function to be plugged into the API endpoint
         /ttype/<tokentype> which is defined in api/ttype.py

privacyidea/lib/tokens/pushtoken.py

@@ -186,9 +186,11 @@ def _build_smartphone_data(serial, challenge, registration_url, pem_privkey, opt
                                                options) or "1"
     sslverify = getParam({"sslverify": sslverify}, "sslverify",
                          allowed_values=["0", "1"], default="1")
-    message_on_mobile = get_action_values_from_options(SCOPE.AUTH,
+    message_on_mobile = str.format(get_action_values_from_options(SCOPE.AUTH,
                                                        PUSH_ACTION.MOBILE_TEXT,
-                                                       options) or DEFAULT_MOBILE_TEXT
+                                                       options) or DEFAULT_MOBILE_TEXT, client_ip=options.get("clientip"),
+                                                       username = options.get('username')
+                                                       )
     title = get_action_values_from_options(SCOPE.AUTH, PUSH_ACTION.MOBILE_TITLE,
                                            options) or "privacyIDEA"
     smartphone_data = {"nonce": challenge,