pyrofork: Implement QrCode Login

wulan17 · wulan17 · commit 40cd7badd53b · 2025-05-14T20:48:31.000+07:00
Signed-off-by: wulan17 &lt;wulan17@komodos.id&gt;
diff --git a/compiler/docs/compiler.py b/compiler/docs/compiler.py
@@ -421,6 +421,7 @@ def get_title_list(s: str) -> list:
             resend_code
             sign_in
             sign_in_bot
+            sign_in_qrcode
             sign_up
             get_password_hint
             check_password
@@ -724,6 +725,7 @@ def get_title_list(s: str) -> list:
         Authorization
             ActiveSession
             ActiveSessions
+            LoginToken
             SentCode
             TermsOfService
         """
diff --git a/pyrogram/client.py b/pyrogram/client.py
@@ -130,6 +130,9 @@ class Client(Methods):
             Pass a session string to load the session in-memory.
             Implies ``in_memory=True``.
 
+        use_qrcode (``bool``, *optional*):
+            Pass True to login using a QR code.
+
         in_memory (``bool``, *optional*):
             Pass True to start an in-memory session that will be discarded as soon as the client stops.
             In order to reconnect again using an in-memory session without having to login again, you can use
@@ -254,6 +257,7 @@ def __init__(
         test_mode: Optional[bool] = False,
         bot_token: Optional[str] = None,
         session_string: Optional[str] = None,
+        use_qrcode: Optional[bool] = False,
         in_memory: Optional[bool] = None,
         mongodb: Optional[dict] = None,
         storage: Optional[Storage] = None,
@@ -289,6 +293,7 @@ def __init__(
         self.test_mode = test_mode
         self.bot_token = bot_token
         self.session_string = session_string
+        self.use_qrcode = use_qrcode
         self.in_memory = in_memory
         self.mongodb = mongodb
         self.phone_number = phone_number
@@ -397,59 +402,76 @@ async def updates_watchdog(self):
             if datetime.now() - self.last_update_time > timedelta(seconds=self.UPDATES_WATCHDOG_INTERVAL):
                 await self.invoke(raw.functions.updates.GetState())
 
+    async def _wait_for_update_login_token(self):
+        """
+        Wait for an UpdateLoginToken update from Telegram.
+        """
+        while True:
+            update, _, _ = await self.dispatcher.updates_queue.get()
+            if isinstance(update, raw.types.UpdateLoginToken):
+                break
+
     async def authorize(self) -> User:
         if self.bot_token:
             return await self.sign_in_bot(self.bot_token)
 
         print(f"Welcome to Pyrogram (version {__version__})")
         print(f"Pyrogram is free software and comes with ABSOLUTELY NO WARRANTY. Licensed\n"
               f"under the terms of the {__license__}.\n")
+        if not self.use_qrcode:
+            while True:
+                try:   
+                    if not self.phone_number:
+                        while True:
+                            print("Enter 'qrcode' if you want to login with qrcode.")
+                            value = await ainput("Enter phone number or bot token: ")
 
-        while True:
-            try:
-                if not self.phone_number:
-                    while True:
-                        value = await ainput("Enter phone number or bot token: ")
+                            if not value:
+                                continue
 
-                        if not value:
-                            continue
+                            if value.lower() == "qrcode":
+                                self.use_qrcode = True
+                                break
 
-                        confirm = (await ainput(f'Is "{value}" correct? (y/N): ')).lower()
+                            confirm = (await ainput(f'Is "{value}" correct? (y/N): ')).lower()
 
-                        if confirm == "y":
-                            break
+                            if confirm == "y":
+                                break
 
-                    if ":" in value:
-                        self.bot_token = value
-                        return await self.sign_in_bot(value)
-                    else:
-                        self.phone_number = value
+                        if ":" in value:
+                            self.bot_token = value
+                            return await self.sign_in_bot(value)
+                        else:
+                            self.phone_number = value
 
-                sent_code = await self.send_code(self.phone_number)
-            except BadRequest as e:
-                print(e.MESSAGE)
-                self.phone_number = None
-                self.bot_token = None
-            else:
-                break
+                    sent_code = await self.send_code(self.phone_number)
+                except BadRequest as e:
+                    print(e.MESSAGE)
+                    self.phone_number = None
+                    self.bot_token = None
+                else:
+                    break
 
-        sent_code_descriptions = {
-            enums.SentCodeType.APP: "Telegram app",
-            enums.SentCodeType.SMS: "SMS",
-            enums.SentCodeType.CALL: "phone call",
-            enums.SentCodeType.FLASH_CALL: "phone flash call",
-            enums.SentCodeType.FRAGMENT_SMS: "Fragment SMS",
-            enums.SentCodeType.EMAIL_CODE: "email code"
-        }
+            sent_code_descriptions = {
+                enums.SentCodeType.APP: "Telegram app",
+                enums.SentCodeType.SMS: "SMS",
+                enums.SentCodeType.CALL: "phone call",
+                enums.SentCodeType.FLASH_CALL: "phone flash call",
+                enums.SentCodeType.FRAGMENT_SMS: "Fragment SMS",
+                enums.SentCodeType.EMAIL_CODE: "email code"
+            }
 
-        print(f"The confirmation code has been sent via {sent_code_descriptions[sent_code.type]}")
+            print(f"The confirmation code has been sent via {sent_code_descriptions[sent_code.type]}")
 
         while True:
-            if not self.phone_code:
+            if not self.use_qrcode and not self.phone_code:
                 self.phone_code = await ainput("Enter confirmation code: ")
 
             try:
-                signed_in = await self.sign_in(self.phone_number, sent_code.phone_code_hash, self.phone_code)
+                if self.use_qrcode:
+                    signed_in = await self.sign_in_qrcode()
+                else:
+                    signed_in = await self.sign_in(self.phone_number, sent_code.phone_code_hash, self.phone_code)
             except BadRequest as e:
                 print(e.MESSAGE)
                 self.phone_code = None
@@ -488,7 +510,15 @@ async def authorize(self) -> User:
                         print(e.MESSAGE)
                         self.password = None
             else:
-                break
+                if self.use_qrcode and isinstance(signed_in, raw.types.auth.LoginToken):
+                    time_out = signed_in.expires - datetime.timestamp(datetime.now())
+                    try:
+                        await asyncio.wait_for(self._wait_for_update_login_token(), timeout=time_out)
+                    except asyncio.TimeoutError:
+                        print("QR code expired, Requesting new Qr code...")
+                    continue
+                else:
+                    break
 
         if isinstance(signed_in, User):
             return signed_in
diff --git a/pyrogram/methods/auth/__init__.py b/pyrogram/methods/auth/__init__.py
@@ -31,6 +31,7 @@
 from .send_recovery_code import SendRecoveryCode
 from .sign_in import SignIn
 from .sign_in_bot import SignInBot
+from .sign_in_qrcode import SignInQrcode
 from .sign_up import SignUp
 from .terminate import Terminate
 
@@ -50,6 +51,7 @@ class Auth(
     SendRecoveryCode,
     SignIn,
     SignInBot,
+    SignInQrcode,
     SignUp,
     Terminate
 ):
diff --git a/pyrogram/methods/auth/sign_in_qrcode.py b/pyrogram/methods/auth/sign_in_qrcode.py
@@ -0,0 +1,139 @@
+#  Pyrofork - Telegram MTProto API Client Library for Python
+#  Copyright (C) 2017-present Dan <https://github.com/delivrance>
+#  Copyright (C) 2022-present Mayuri-Chan <https://github.com/Mayuri-Chan>
+#
+#  This file is part of Pyrofork.
+#
+#  Pyrofork is free software: you can redistribute it and/or modify
+#  it under the terms of the GNU Lesser General Public License as published
+#  by the Free Software Foundation, either version 3 of the License, or
+#  (at your option) any later version.
+#
+#  Pyrofork is distributed in the hope that it will be useful,
+#  but WITHOUT ANY WARRANTY; without even the implied warranty of
+#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#  GNU Lesser General Public License for more details.
+#
+#  You should have received a copy of the GNU Lesser General Public License
+#  along with Pyrofork.  If not, see <http://www.gnu.org/licenses/>.
+
+import logging
+from base64 import b64encode
+from typing import Union
+
+import pyrogram
+from pyrogram import raw
+from pyrogram import types
+from pyrogram.session import Session, Auth
+
+log = logging.getLogger(__name__)
+
+QRCODE_AVAIL = False
+try:
+    import qrcode
+    QRCODE_AVAIL = True
+except ImportError:
+    QRCODE_AVAIL = False
+
+
+class SignInQrcode:
+    async def sign_in_qrcode(
+        self: "pyrogram.Client"
+    ) -> Union["types.User", "types.LoginToken"]:
+        """Authorize a user in Telegram with a QR code.
+
+        .. include:: /_includes/usable-by/users.rst
+
+        Returns:
+            :obj:`~pyrogram.types.User` | :obj:`pyrogram.types.LoginToken`, in case the
+            authorization completed, the user is returned. In case the QR code is
+            not scanned, a login token is returned.
+
+        Raises:
+            ImportError: In case the qrcode library is not installed.
+            SessionPasswordNeeded: In case a password is needed to sign in.
+        """
+
+        if not QRCODE_AVAIL:
+            raise ImportError("qrcode is missing! "
+                            "Please install it with `pip install qrcode`")
+        r = await self.session.invoke(
+            raw.functions.auth.ExportLoginToken(
+                api_id=self.api_id,
+                api_hash=self.api_hash,
+                except_ids=[]
+            )
+        )
+        if isinstance(r, raw.types.auth.LoginToken):
+            base64_token = b64encode(r.token).decode("utf-8")
+            login_url = f"tg://login?token={base64_token}"
+            qr = qrcode.QRCode(
+                version=1,
+                error_correction=qrcode.constants.ERROR_CORRECT_L,
+                box_size=10,
+                border=4,
+            )
+            qr.add_data(login_url)
+            qr.make(fit=True)
+
+            print("Scan the QR code with your Telegram app.")
+            qr.print_ascii()
+
+            return r
+        elif isinstance(r, raw.types.auth.LoginTokenSuccess):
+            await self.storage.user_id(r.authorization.user.id)
+            await self.storage.is_bot(False)
+
+            return types.User._parse(self, r.authorization.user)
+        elif isinstance(r, raw.types.auth.LoginTokenMigrateTo):
+            # pylint: disable=access-member-before-definition
+            await self.session.stop()
+
+            await self.storage.dc_id(r.dc_id)
+            await self.storage.auth_key(
+                await Auth(
+                    self, await self.storage.dc_id(),
+                    await self.storage.test_mode()
+                ).create()
+            )
+            self.session = Session(
+                self, await self.storage.dc_id(),
+                await self.storage.auth_key(), await self.storage.test_mode()
+            )
+
+            await self.session.start()
+            r = await self.session.invoke(
+                raw.functions.auth.ImportLoginToken(
+                    token=r.token
+                )
+            )
+            if isinstance(r, raw.types.auth.LoginToken):
+                base64_token = b64encode(r.token).decode("utf-8")
+                login_url = f"tg://login?token={base64_token}"
+                qr = qrcode.QRCode(
+                    version=1,
+                    error_correction=qrcode.constants.ERROR_CORRECT_L,
+                    box_size=10,
+                    border=4,
+                )
+                qr.add_data(login_url)
+                qr.make(fit=True)
+
+                print("Scan the QR code below with your Telegram app.")
+                qr.print_ascii()
+
+                return types.LoginToken(
+                    self,
+                    r.token,
+                    r.expires
+                )
+            elif isinstance(r, raw.types.auth.LoginTokenSuccess):
+                await self.storage.user_id(r.authorization.user.id)
+                await self.storage.is_bot(False)
+
+                return types.User._parse(self, r.authorization.user)
+        else:
+            raise pyrogram.exceptions.RPCError(
+                "Unknown response type from Telegram API"
+            )
+        return r
diff --git a/pyrogram/types/authorization/__init__.py b/pyrogram/types/authorization/__init__.py
@@ -19,12 +19,14 @@
 
 from .active_session import ActiveSession
 from .active_sessions import ActiveSessions
+from .login_token import LoginToken
 from .sent_code import SentCode
 from .terms_of_service import TermsOfService
 
 __all__ = [
     "ActiveSession",
     "ActiveSessions",
+    "LoginToken",
     "SentCode",
     "TermsOfService",
 ]
diff --git a/pyrogram/types/authorization/login_token.py b/pyrogram/types/authorization/login_token.py
@@ -0,0 +1,37 @@
+#  Pyrofork - Telegram MTProto API Client Library for Python
+#  Copyright (C) 2022-present Mayuri-Chan <https://github.com/Mayuri-Chan>
+#
+#  This file is part of Pyrofork.
+#
+#  Pyrofork is free software: you can redistribute it and/or modify
+#  it under the terms of the GNU Lesser General Public License as published
+#  by the Free Software Foundation, either version 3 of the License, or
+#  (at your option) any later version.
+#
+#  Pyrofork is distributed in the hope that it will be useful,
+#  but WITHOUT ANY WARRANTY; without even the implied warranty of
+#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#  GNU Lesser General Public License for more details.
+#
+#  You should have received a copy of the GNU Lesser General Public License
+#  along with Pyrofork.  If not, see <http://www.gnu.org/licenses/>.
+
+from ..object import Object
+
+
+class LoginToken(Object):
+    """Contains info on a login token.
+
+    Parameters:
+        token (``str``):
+            The login token.
+
+        expires (``int``):
+            The expiration date of the token in UNIX format.
+    """
+
+    def __init__(self, *, token: str, expires: int):
+        super().__init__()
+
+        self.token = token
+        self.expires = expires
