Add a friendly method for QR login

Closes #1471.
This commit is contained in:
Lonami Exo 2020-06-05 21:58:59 +02:00
parent bfa995d52b
commit c904b7ccd8
3 changed files with 166 additions and 0 deletions

View File

@ -136,6 +136,15 @@ MessageButton
:show-inheritance:
QRLogin
=======
.. automodule:: telethon.qrlogin
:members:
:undoc-members:
:show-inheritance:
SenderGetter
============

View File

@ -5,6 +5,7 @@ import sys
import typing
from .. import utils, helpers, errors, password as pwd_mod
from ..qrlogin import QRLogin
from ..tl import types, functions
if typing.TYPE_CHECKING:
@ -496,6 +497,43 @@ class AuthMethods:
return result
async def qr_login(self: 'TelegramClient', ignored_ids: typing.List[int] = None) -> QRLogin:
"""
Initiates the QR login procedure.
Note that you must be connected before invoking this, as with any
other request.
It is up to the caller to decide how to present the code to the user,
whether it's the URL, using the token bytes directly, or generating
a QR code and displaying it by other means.
See the documentation for `QRLogin` to see how to proceed after this.
Arguments
ignored_ids (List[`int`]):
List of already logged-in user IDs, to prevent logging in
twice with the same user.
Returns
An instance of `QRLogin`.
Example
.. code-block:: python
def display_url_as_qr(url):
pass # do whatever to show url as a qr to the user
qr_login = await client.qr_login()
display_url_as_qr(qr_login.url)
# Important! You need to wait for the login to complete!
await qr_login.wait()
"""
qr_login = QRLogin(self, ignored_ids or [])
await qr_login.recreate()
return qr_login
async def log_out(self: 'TelegramClient') -> bool:
"""
Logs out Telegram and deletes the current ``*.session`` file.

119
telethon/qrlogin.py Normal file
View File

@ -0,0 +1,119 @@
import asyncio
import base64
import datetime
from telethon import events
from telethon.tl import types, functions
class QRLogin:
"""
QR login information.
Most of the time, you will present the `url` as a QR code to the user,
and while it's being shown, call `wait`.
"""
def __init__(self, client, ignored_ids):
self._client = client
self._request = functions.auth.ExportLoginTokenRequest(
self._client.api_id, self._client.api_hash, ignored_ids)
self._resp = None
async def recreate(self):
"""
Generates a new token and URL for a new QR code, useful if the code
has expired before it was imported.
"""
self._resp = await self._client(self._request)
@property
def token(self) -> bytes:
"""
The binary data representing the token.
It can be used by a previously-authorized client in a call to
:tl:`auth.importLoginToken` to log the client that originally
requested the QR login.
"""
return self._resp.token
@property
def url(self) -> str:
"""
The ``tg://login`` URI with the token. When opened by a Telegram
application where the user is logged in, it will import the login
token.
If you want to display a QR code to the user, this is the URL that
should be launched when the QR code is scanned (the URL that should
be contained in the QR code image you generate).
Whether you generate the QR code image or not is up to you, and the
library can't do this for you due to the vast ways of generating and
displaying the QR code that exist.
The URL simply consists of `token` base64-encoded.
"""
return 'tg://login?token={}'.format(base64.b64encode(self._resp.token))
@property
def expires(self) -> datetime.datetime:
"""
The `datetime` at which the QR code will expire.
If you want to try again, you will need to call `recreate`.
"""
return self._resp.expires
async def wait(self, timeout: float = None):
"""
Waits for the token to be imported by a previously-authorized client,
either by scanning the QR, launching the URL directly, or calling the
import method.
This method **must** be called before the QR code is scanned, and
must be executing while the QR code is being scanned. Otherwise, the
login will not complete.
Will raise `asyncio.TimeoutError` if the login doesn't complete on
time.
Arguments
timeout (float):
The timeout, in seconds, to wait before giving up. By default
the library will wait until the token expires, which is often
what you want.
Returns
On success, an instance of :tl:`User`. On failure it will raise.
"""
if timeout is None:
timeout = (self._resp.expires - datetime.datetime.now(tz=datetime.timezone.utc)).total_seconds()
event = asyncio.Event()
async def handler(_update):
event.set()
self._client.add_event_handler(handler, events.Raw(types.UpdateLoginToken))
try:
# Will raise timeout error if it doesn't complete quick enough,
# which we want to let propagate
await asyncio.wait_for(event.wait(), timeout=timeout)
finally:
self._client.remove_event_handler(handler)
# We got here without it raising timeout error, so we can proceed
resp = await self._client(self._request)
if isinstance(resp, types.auth.LoginTokenMigrateTo):
await self._client._switch_dc(resp.dc_id)
resp = await self._client(functions.auth.ImportLoginTokenRequest(resp.token))
# resp should now be auth.loginTokenSuccess
if isinstance(resp, types.auth.LoginTokenSuccess):
user = resp.authorization.user
self._client._on_login(user)
return user
raise TypeError('Login token response was unexpected: {}'.format(resp))