From ac2e59b472f4b6eb8cbd744f4d5bda0c0d0d0b6a Mon Sep 17 00:00:00 2001 From: Lonami Exo Date: Sun, 10 Jun 2018 12:57:36 +0200 Subject: [PATCH] Separate auth requests from the TelegramClient --- telethon/client/auth.py | 424 ++++++++++++++++++++++++++++++ telethon/client/telegramclient.py | 424 +----------------------------- 2 files changed, 425 insertions(+), 423 deletions(-) create mode 100644 telethon/client/auth.py diff --git a/telethon/client/auth.py b/telethon/client/auth.py new file mode 100644 index 00000000..10913ad6 --- /dev/null +++ b/telethon/client/auth.py @@ -0,0 +1,424 @@ +import getpass +import hashlib +import sys + +import os + +from .messageparse import MessageParseMethods +from .users import UserMethods +from .. import utils, helpers, errors +from ..tl import types, functions + + +class AuthMethods(MessageParseMethods, UserMethods): + + # region Public methods + + async def start( + self, + phone=lambda: input('Please enter your phone: '), + password=lambda: getpass.getpass('Please enter your password: '), + bot_token=None, force_sms=False, code_callback=None, + first_name='New User', last_name=''): + """ + Convenience method to interactively connect and sign in if required, + also taking into consideration that 2FA may be enabled in the account. + + If the phone doesn't belong to an existing account (and will hence + `sign_up` for a new one), **you are agreeing to Telegram's + Terms of Service. This is required and your account + will be banned otherwise.** See https://telegram.org/tos + and https://core.telegram.org/api/terms. + + Example usage: + >>> client = ... + >>> client.start(phone) + Please enter the code you received: 12345 + Please enter your password: ******* + (You are now logged in) + + Args: + phone (`str` | `int` | `callable`): + The phone (or callable without arguments to get it) + to which the code will be sent. + + password (`callable`, optional): + The password for 2 Factor Authentication (2FA). + This is only required if it is enabled in your account. + + bot_token (`str`): + Bot Token obtained by `@BotFather `_ + to log in as a bot. Cannot be specified with ``phone`` (only + one of either allowed). + + force_sms (`bool`, optional): + Whether to force sending the code request as SMS. + This only makes sense when signing in with a `phone`. + + code_callback (`callable`, optional): + A callable that will be used to retrieve the Telegram + login code. Defaults to `input()`. + + first_name (`str`, optional): + The first name to be used if signing up. This has no + effect if the account already exists and you sign in. + + last_name (`str`, optional): + Similar to the first name, but for the last. Optional. + + Returns: + This `TelegramClient`, so initialization + can be chained with ``.start()``. + """ + + if code_callback is None: + def code_callback(): + return input('Please enter the code you received: ') + elif not callable(code_callback): + raise ValueError( + 'The code_callback parameter needs to be a callable ' + 'function that returns the code you received by Telegram.' + ) + + if not phone and not bot_token: + raise ValueError('No phone number or bot token provided.') + + if phone and bot_token and not callable(phone): + raise ValueError('Both a phone and a bot token provided, ' + 'must only provide one of either') + + if not self.is_connected(): + await self.connect() + + if await self.is_user_authorized(): + return self + + if bot_token: + await self.sign_in(bot_token=bot_token) + return self + + # Turn the callable into a valid phone number + while callable(phone): + phone = utils.parse_phone(phone()) or phone + + me = None + attempts = 0 + max_attempts = 3 + two_step_detected = False + + sent_code = await self.send_code_request(phone, force_sms=force_sms) + sign_up = not sent_code.phone_registered + while attempts < max_attempts: + try: + if sign_up: + me = await self.sign_up( + code_callback(), first_name, last_name) + else: + # Raises SessionPasswordNeededError if 2FA enabled + me = await self.sign_in(phone, code_callback()) + break + except errors.SessionPasswordNeededError: + two_step_detected = True + break + except errors.PhoneNumberOccupiedError: + sign_up = False + except errors.PhoneNumberUnoccupiedError: + sign_up = True + except (errors.PhoneCodeEmptyError, + errors.PhoneCodeExpiredError, + errors.PhoneCodeHashEmptyError, + errors.PhoneCodeInvalidError): + print('Invalid code. Please try again.', file=sys.stderr) + + attempts += 1 + else: + raise RuntimeError( + '{} consecutive sign-in attempts failed. Aborting' + .format(max_attempts) + ) + + if two_step_detected: + if not password: + raise ValueError( + "Two-step verification is enabled for this account. " + "Please provide the 'password' argument to 'start()'." + ) + # TODO If callable given make it retry on invalid + if callable(password): + password = password() + me = await self.sign_in(phone=phone, password=password) + + # We won't reach here if any step failed (exit by exception) + signed, name = 'Signed in successfully as', utils.get_display_name(me) + try: + print(signed, name) + except UnicodeEncodeError: + # Some terminals don't support certain characters + print(signed, name.encode('utf-8', errors='ignore') + .decode('ascii', errors='ignore')) + + return self + + async def is_user_authorized(self): + return await self.get_me() is not None + + async def sign_in( + self, phone=None, code=None, password=None, + bot_token=None, phone_code_hash=None): + """ + Starts or completes the sign in process with the given phone number + or code that Telegram sent. + + Args: + phone (`str` | `int`): + The phone to send the code to if no code was provided, + or to override the phone that was previously used with + these requests. + + code (`str` | `int`): + The code that Telegram sent. Note that if you have sent this + code through the application itself it will immediately + expire. If you want to send the code, obfuscate it somehow. + If you're not doing any of this you can ignore this note. + + password (`str`): + 2FA password, should be used if a previous call raised + SessionPasswordNeededError. + + bot_token (`str`): + Used to sign in as a bot. Not all requests will be available. + This should be the hash the @BotFather gave you. + + phone_code_hash (`str`): + The hash returned by .send_code_request. This can be set to None + to use the last hash known. + + Returns: + The signed in user, or the information about + :meth:`send_code_request`. + """ + me = await self.get_me() + if me: + return me + + if phone and not code and not password: + return await self.send_code_request(phone) + elif code: + phone = utils.parse_phone(phone) or self._phone + phone_code_hash = \ + phone_code_hash or self._phone_code_hash.get(phone, None) + + if not phone: + raise ValueError( + 'Please make sure to call send_code_request first.' + ) + if not phone_code_hash: + raise ValueError('You also need to provide a phone_code_hash.') + + # May raise PhoneCodeEmptyError, PhoneCodeExpiredError, + # PhoneCodeHashEmptyError or PhoneCodeInvalidError. + result = await self(functions.auth.SignInRequest( + phone, phone_code_hash, str(code))) + elif password: + salt = (await self( + functions.account.GetPasswordRequest())).current_salt + result = await self(functions.auth.CheckPasswordRequest( + helpers.get_password_hash(password, salt) + )) + elif bot_token: + result = await self(functions.auth.ImportBotAuthorizationRequest( + flags=0, bot_auth_token=bot_token, + api_id=self.api_id, api_hash=self.api_hash + )) + else: + raise ValueError( + 'You must provide a phone and a code the first time, ' + 'and a password only if an RPCError was raised before.' + ) + + self._self_input_peer = utils.get_input_peer( + result.user, allow_self=False + ) + return result.user + + async def sign_up(self, code, first_name, last_name=''): + """ + Signs up to Telegram if you don't have an account yet. + You must call .send_code_request(phone) first. + + **By using this method you're agreeing to Telegram's + Terms of Service. This is required and your account + will be banned otherwise.** See https://telegram.org/tos + and https://core.telegram.org/api/terms. + + Args: + code (`str` | `int`): + The code sent by Telegram + + first_name (`str`): + The first name to be used by the new account. + + last_name (`str`, optional) + Optional last name. + + Returns: + The new created :tl:`User`. + """ + me = await self.get_me() + if me: + return me + + if self._tos and self._tos.text: + if self.parse_mode: + t = self.parse_mode.unparse(self._tos.text, self._tos.entities) + else: + t = self._tos.text + sys.stderr.write("{}\n".format(t)) + sys.stderr.flush() + + result = await self(functions.auth.SignUpRequest( + phone_number=self._phone, + phone_code_hash=self._phone_code_hash.get(self._phone, ''), + phone_code=str(code), + first_name=first_name, + last_name=last_name + )) + + if self._tos: + await self( + functions.help.AcceptTermsOfServiceRequest(self._tos.id)) + + self._self_input_peer = utils.get_input_peer( + result.user, allow_self=False + ) + return result.user + + async def send_code_request(self, phone, force_sms=False): + """ + Sends a code request to the specified phone number. + + Args: + phone (`str` | `int`): + The phone to which the code will be sent. + + force_sms (`bool`, optional): + Whether to force sending as SMS. + + Returns: + An instance of :tl:`SentCode`. + """ + phone = utils.parse_phone(phone) or self._phone + phone_hash = self._phone_code_hash.get(phone) + + if not phone_hash: + result = await self(functions.auth.SendCodeRequest( + phone, self.api_id, self.api_hash)) + self._tos = result.terms_of_service + self._phone_code_hash[phone] = phone_hash = result.phone_code_hash + else: + force_sms = True + + self._phone = phone + + if force_sms: + result = await self( + functions.auth.ResendCodeRequest(phone, phone_hash)) + + self._phone_code_hash[phone] = result.phone_code_hash + + return result + + async def log_out(self): + """ + Logs out Telegram and deletes the current ``*.session`` file. + + Returns: + ``True`` if the operation was successful. + """ + try: + await self(functions.auth.LogOutRequest()) + except errors.RPCError: + return False + + await self.disconnect() + self.session.delete() + self._authorized = False + return True + + async def edit_2fa( + self, current_password=None, new_password=None, hint='', + email=None): + """ + Changes the 2FA settings of the logged in user, according to the + passed parameters. Take note of the parameter explanations. + + Has no effect if both current and new password are omitted. + + current_password (`str`, optional): + The current password, to authorize changing to ``new_password``. + Must be set if changing existing 2FA settings. + Must **not** be set if 2FA is currently disabled. + Passing this by itself will remove 2FA (if correct). + + new_password (`str`, optional): + The password to set as 2FA. + If 2FA was already enabled, ``current_password`` **must** be set. + Leaving this blank or ``None`` will remove the password. + + hint (`str`, optional): + Hint to be displayed by Telegram when it asks for 2FA. + Leaving unspecified is highly discouraged. + Has no effect if ``new_password`` is not set. + + email (`str`, optional): + Recovery and verification email. Raises ``EmailUnconfirmedError`` + if value differs from current one, and has no effect if + ``new_password`` is not set. + + Returns: + ``True`` if successful, ``False`` otherwise. + """ + if new_password is None and current_password is None: + return False + + pass_result = await self(functions.account.GetPasswordRequest()) + if isinstance( + pass_result, types.account.NoPassword) and current_password: + current_password = None + + salt_random = os.urandom(8) + salt = pass_result.new_salt + salt_random + if not current_password: + current_password_hash = salt + else: + current_password = ( + pass_result.current_salt + + current_password.encode() + + pass_result.current_salt + ) + current_password_hash = hashlib.sha256(current_password).digest() + + if new_password: # Setting new password + new_password = salt + new_password.encode('utf-8') + salt + new_password_hash = hashlib.sha256(new_password).digest() + new_settings = types.account.PasswordInputSettings( + new_salt=salt, + new_password_hash=new_password_hash, + hint=hint + ) + if email: # If enabling 2FA or changing email + new_settings.email = email # TG counts empty string as None + return await self(functions.account.UpdatePasswordSettingsRequest( + current_password_hash, new_settings=new_settings + )) + else: # Removing existing password + return await self(functions.account.UpdatePasswordSettingsRequest( + current_password_hash, + new_settings=types.account.PasswordInputSettings( + new_salt=bytes(), + new_password_hash=bytes(), + hint=hint + ) + )) + + # endregion diff --git a/telethon/client/telegramclient.py b/telethon/client/telegramclient.py index 30352217..fbe453ee 100644 --- a/telethon/client/telegramclient.py +++ b/telethon/client/telegramclient.py @@ -1,10 +1,6 @@ -import getpass -import hashlib import logging -import sys import warnings -from ..tl.functions.help import AcceptTermsOfServiceRequest from ..tl.functions.updates import GetDifferenceRequest from ..tl.types.updates import ( DifferenceSlice, DifferenceEmpty, Difference, DifferenceTooLong @@ -17,30 +13,13 @@ except ImportError: from .telegrambaseclient import TelegramBaseClient -from .. import helpers, events -from ..errors import ( - PhoneCodeEmptyError, PhoneCodeExpiredError, - PhoneCodeHashEmptyError, PhoneCodeInvalidError, SessionPasswordNeededError, - PhoneNumberUnoccupiedError, - PhoneNumberOccupiedError -) -from ..tl.functions.account import ( - GetPasswordRequest, UpdatePasswordSettingsRequest -) -from ..tl.functions.auth import ( - CheckPasswordRequest, LogOutRequest, SendCodeRequest, SignInRequest, - SignUpRequest, ResendCodeRequest, ImportBotAuthorizationRequest -) +from .. import events from ..tl.types import ( UpdateNewMessage, Updates ) -from ..tl.types.account import PasswordInputSettings, NoPassword __log__ = logging.getLogger(__name__) -import os -from .. import utils -from ..errors import RPCError class TelegramClient(TelegramBaseClient): @@ -53,336 +32,6 @@ class TelegramClient(TelegramBaseClient): # region Telegram requests functions - # region Authorization requests - - def send_code_request(self, phone, force_sms=False): - """ - Sends a code request to the specified phone number. - - Args: - phone (`str` | `int`): - The phone to which the code will be sent. - - force_sms (`bool`, optional): - Whether to force sending as SMS. - - Returns: - An instance of :tl:`SentCode`. - """ - phone = utils.parse_phone(phone) or self._phone - phone_hash = self._phone_code_hash.get(phone) - - if not phone_hash: - result = self(SendCodeRequest(phone, self.api_id, self.api_hash)) - self._tos = result.terms_of_service - self._phone_code_hash[phone] = phone_hash = result.phone_code_hash - else: - force_sms = True - - self._phone = phone - - if force_sms: - result = self(ResendCodeRequest(phone, phone_hash)) - self._phone_code_hash[phone] = result.phone_code_hash - - return result - - def start(self, - phone=lambda: input('Please enter your phone: '), - password=lambda: getpass.getpass('Please enter your password: '), - bot_token=None, force_sms=False, code_callback=None, - first_name='New User', last_name=''): - """ - Convenience method to interactively connect and sign in if required, - also taking into consideration that 2FA may be enabled in the account. - - If the phone doesn't belong to an existing account (and will hence - `sign_up` for a new one), **you are agreeing to Telegram's - Terms of Service. This is required and your account - will be banned otherwise.** See https://telegram.org/tos - and https://core.telegram.org/api/terms. - - Example usage: - >>> client = TelegramClient(session, api_id, api_hash).start(phone) - Please enter the code you received: 12345 - Please enter your password: ******* - (You are now logged in) - - Args: - phone (`str` | `int` | `callable`): - The phone (or callable without arguments to get it) - to which the code will be sent. - - password (`callable`, optional): - The password for 2 Factor Authentication (2FA). - This is only required if it is enabled in your account. - - bot_token (`str`): - Bot Token obtained by `@BotFather `_ - to log in as a bot. Cannot be specified with ``phone`` (only - one of either allowed). - - force_sms (`bool`, optional): - Whether to force sending the code request as SMS. - This only makes sense when signing in with a `phone`. - - code_callback (`callable`, optional): - A callable that will be used to retrieve the Telegram - login code. Defaults to `input()`. - - first_name (`str`, optional): - The first name to be used if signing up. This has no - effect if the account already exists and you sign in. - - last_name (`str`, optional): - Similar to the first name, but for the last. Optional. - - Returns: - This `TelegramClient`, so initialization - can be chained with ``.start()``. - """ - - if code_callback is None: - def code_callback(): - return input('Please enter the code you received: ') - elif not callable(code_callback): - raise ValueError( - 'The code_callback parameter needs to be a callable ' - 'function that returns the code you received by Telegram.' - ) - - if not phone and not bot_token: - raise ValueError('No phone number or bot token provided.') - - if phone and bot_token and not callable(phone): - raise ValueError('Both a phone and a bot token provided, ' - 'must only provide one of either') - - if not self.is_connected(): - self.connect() - - if self.is_user_authorized(): - self._check_events_pending_resolve() - return self - - if bot_token: - self.sign_in(bot_token=bot_token) - return self - - # Turn the callable into a valid phone number - while callable(phone): - phone = utils.parse_phone(phone()) or phone - - me = None - attempts = 0 - max_attempts = 3 - two_step_detected = False - - sent_code = self.send_code_request(phone, force_sms=force_sms) - sign_up = not sent_code.phone_registered - while attempts < max_attempts: - try: - if sign_up: - me = self.sign_up(code_callback(), first_name, last_name) - else: - # Raises SessionPasswordNeededError if 2FA enabled - me = self.sign_in(phone, code_callback()) - break - except SessionPasswordNeededError: - two_step_detected = True - break - except PhoneNumberOccupiedError: - sign_up = False - except PhoneNumberUnoccupiedError: - sign_up = True - except (PhoneCodeEmptyError, PhoneCodeExpiredError, - PhoneCodeHashEmptyError, PhoneCodeInvalidError): - print('Invalid code. Please try again.', file=sys.stderr) - - attempts += 1 - else: - raise RuntimeError( - '{} consecutive sign-in attempts failed. Aborting' - .format(max_attempts) - ) - - if two_step_detected: - if not password: - raise ValueError( - "Two-step verification is enabled for this account. " - "Please provide the 'password' argument to 'start()'." - ) - # TODO If callable given make it retry on invalid - if callable(password): - password = password() - me = self.sign_in(phone=phone, password=password) - - # We won't reach here if any step failed (exit by exception) - signed, name = 'Signed in successfully as', utils.get_display_name(me) - try: - print(signed, name) - except UnicodeEncodeError: - # Some terminals don't support certain characters - print(signed, name.encode('utf-8', errors='ignore') - .decode('ascii', errors='ignore')) - - self._check_events_pending_resolve() - return self - - def sign_in(self, phone=None, code=None, - password=None, bot_token=None, phone_code_hash=None): - """ - Starts or completes the sign in process with the given phone number - or code that Telegram sent. - - Args: - phone (`str` | `int`): - The phone to send the code to if no code was provided, - or to override the phone that was previously used with - these requests. - - code (`str` | `int`): - The code that Telegram sent. Note that if you have sent this - code through the application itself it will immediately - expire. If you want to send the code, obfuscate it somehow. - If you're not doing any of this you can ignore this note. - - password (`str`): - 2FA password, should be used if a previous call raised - SessionPasswordNeededError. - - bot_token (`str`): - Used to sign in as a bot. Not all requests will be available. - This should be the hash the @BotFather gave you. - - phone_code_hash (`str`): - The hash returned by .send_code_request. This can be set to None - to use the last hash known. - - Returns: - The signed in user, or the information about - :meth:`send_code_request`. - """ - if self.is_user_authorized(): - self._check_events_pending_resolve() - return self.get_me() - - if phone and not code and not password: - return self.send_code_request(phone) - elif code: - phone = utils.parse_phone(phone) or self._phone - phone_code_hash = \ - phone_code_hash or self._phone_code_hash.get(phone, None) - - if not phone: - raise ValueError( - 'Please make sure to call send_code_request first.' - ) - if not phone_code_hash: - raise ValueError('You also need to provide a phone_code_hash.') - - # May raise PhoneCodeEmptyError, PhoneCodeExpiredError, - # PhoneCodeHashEmptyError or PhoneCodeInvalidError. - result = self(SignInRequest(phone, phone_code_hash, str(code))) - elif password: - salt = self(GetPasswordRequest()).current_salt - result = self(CheckPasswordRequest( - helpers.get_password_hash(password, salt) - )) - elif bot_token: - result = self(ImportBotAuthorizationRequest( - flags=0, bot_auth_token=bot_token, - api_id=self.api_id, api_hash=self.api_hash - )) - else: - raise ValueError( - 'You must provide a phone and a code the first time, ' - 'and a password only if an RPCError was raised before.' - ) - - self._self_input_peer = utils.get_input_peer( - result.user, allow_self=False - ) - self._set_connected_and_authorized() - return result.user - - def sign_up(self, code, first_name, last_name=''): - """ - Signs up to Telegram if you don't have an account yet. - You must call .send_code_request(phone) first. - - **By using this method you're agreeing to Telegram's - Terms of Service. This is required and your account - will be banned otherwise.** See https://telegram.org/tos - and https://core.telegram.org/api/terms. - - Args: - code (`str` | `int`): - The code sent by Telegram - - first_name (`str`): - The first name to be used by the new account. - - last_name (`str`, optional) - Optional last name. - - Returns: - The new created :tl:`User`. - """ - if self.is_user_authorized(): - self._check_events_pending_resolve() - return self.get_me() - - if self._tos and self._tos.text: - if self.parse_mode: - t = self.parse_mode.unparse(self._tos.text, self._tos.entities) - else: - t = self._tos.text - sys.stderr.write("{}\n".format(t)) - sys.stderr.flush() - - result = self(SignUpRequest( - phone_number=self._phone, - phone_code_hash=self._phone_code_hash.get(self._phone, ''), - phone_code=str(code), - first_name=first_name, - last_name=last_name - )) - - if self._tos: - self(AcceptTermsOfServiceRequest(self._tos.id)) - - self._self_input_peer = utils.get_input_peer( - result.user, allow_self=False - ) - self._set_connected_and_authorized() - return result.user - - def log_out(self): - """ - Logs out Telegram and deletes the current ``*.session`` file. - - Returns: - ``True`` if the operation was successful. - """ - try: - self(LogOutRequest()) - except RPCError: - return False - - self.disconnect() - self.session.delete() - self._authorized = False - return True - - # endregion - - # region Downloading media requests - - # endregion - - # endregion - # region Event handling def on(self, event): @@ -554,75 +203,4 @@ class TelegramClient(TelegramBaseClient): super()._set_connected_and_authorized() self._check_events_pending_resolve() - def edit_2fa(self, current_password=None, new_password=None, hint='', - email=None): - """ - Changes the 2FA settings of the logged in user, according to the - passed parameters. Take note of the parameter explanations. - - Has no effect if both current and new password are omitted. - - current_password (`str`, optional): - The current password, to authorize changing to ``new_password``. - Must be set if changing existing 2FA settings. - Must **not** be set if 2FA is currently disabled. - Passing this by itself will remove 2FA (if correct). - - new_password (`str`, optional): - The password to set as 2FA. - If 2FA was already enabled, ``current_password`` **must** be set. - Leaving this blank or ``None`` will remove the password. - - hint (`str`, optional): - Hint to be displayed by Telegram when it asks for 2FA. - Leaving unspecified is highly discouraged. - Has no effect if ``new_password`` is not set. - - email (`str`, optional): - Recovery and verification email. Raises ``EmailUnconfirmedError`` - if value differs from current one, and has no effect if - ``new_password`` is not set. - - Returns: - ``True`` if successful, ``False`` otherwise. - """ - if new_password is None and current_password is None: - return False - - pass_result = self(GetPasswordRequest()) - if isinstance(pass_result, NoPassword) and current_password: - current_password = None - - salt_random = os.urandom(8) - salt = pass_result.new_salt + salt_random - if not current_password: - current_password_hash = salt - else: - current_password = pass_result.current_salt +\ - current_password.encode() + pass_result.current_salt - current_password_hash = hashlib.sha256(current_password).digest() - - if new_password: # Setting new password - new_password = salt + new_password.encode('utf-8') + salt - new_password_hash = hashlib.sha256(new_password).digest() - new_settings = PasswordInputSettings( - new_salt=salt, - new_password_hash=new_password_hash, - hint=hint - ) - if email: # If enabling 2FA or changing email - new_settings.email = email # TG counts empty string as None - return self(UpdatePasswordSettingsRequest( - current_password_hash, new_settings=new_settings - )) - else: # Removing existing password - return self(UpdatePasswordSettingsRequest( - current_password_hash, - new_settings=PasswordInputSettings( - new_salt=bytes(), - new_password_hash=bytes(), - hint=hint - ) - )) - # endregion