LonamiWebs/Telethon
1
1
Fork 0
mirror of https://github.com/LonamiWebs/Telethon.git synced 2024-11-25 10:53:44 +03:00
Code Issues Packages Projects Releases Wiki Activity

Make lint happier

Browse Source
This commit is contained in:
Lonami Exo 2017-05-21 13:59:16 +02:00
parent 63c89af983
commit 02a847b64a
20 changed files with 92 additions and 84 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
README.rst
View File

@ -70,10 +70,10 @@ If you've installed Telethon via pip, launch an interactive python3 session and
.. code:: python
>>> from telethon import InteractiveTelegramClient
>>> # 'sessionid' can be 'yourname'. It'll be saved as yourname.session
>>> # 'session_id' can be 'your_name'. It'll be saved as your_name.session
>>> # Also (obviously) replace the api_id and api_hash with your values
...
>>> client = InteractiveTelegramClient('sessionid', '+34600000000',
>>> client = InteractiveTelegramClient('session_id', '+34600000000',
... api_id=12345, api_hash='0123456789abcdef0123456789abcdef')
==================
@ -93,7 +93,7 @@ Then, simply run ``python3 try_telethon.py`` to start the interactive example.
Using Telethon
==============
If you really want to learn how to use Telethon, it is **highly adviced** that
If you really want to learn how to use Telethon, it is **highly advised** that
you take a look to the
`InteractiveTelegramClient <telethon/interactive_telegram_client.py>`_ file
and check how it works. This file contains everything you'll need to build
@ -208,7 +208,7 @@ Once this is done, pass the proxy settings to the ``TelegramClient`` constructor
>>> from telethon import InteractiveTelegramClient
>>> import socks
>>> client = InteractiveTelegramClient('sessionid', '+34600000000',
>>> client = InteractiveTelegramClient('session_id', '+34600000000',
... api_id=12345, api_hash='0123456789abcdef0123456789abcdef',
... proxy=(socks.SOCKS5, 'localhost', 4444))

20
docs/generate.py
View File

@ -63,20 +63,20 @@ def get_create_path_for(tlobject):
return os.path.join(out_dir, get_file_name(tlobject, add_extension=True))
def get_path_for_type(type, relative_to='.'):
def get_path_for_type(type_, relative_to='.'):
"""Similar to getting the path for a TLObject, it might not be possible
to have the TLObject itself but rather its name (the type);
this method works in the same way, returning a relative path"""
if type.lower() in {'int', 'long', 'int128', 'int256', 'double',
if type_.lower() in {'int', 'long', 'int128', 'int256', 'double',
'vector', 'string', 'bool', 'true', 'bytes', 'date'}:
path = 'index.html#%s' % type.lower()
path = 'index.html#%s' % type_.lower()
elif '.' in type:
elif '.' in type_:
# If it's not a core type, then it has to be a custom Telegram type
namespace, name = type.split('.')
namespace, name = type_.split('.')
path = 'types/%s/%s' % (namespace, get_file_name(name, add_extension=True))
else:
path = 'types/%s' % get_file_name(type, add_extension=True)
path = 'types/%s' % get_file_name(type_, add_extension=True)
return get_relative_path(path, relative_to)
@ -128,7 +128,7 @@ def build_menu(docs, filename, relative_main_index):
def generate_index(folder, original_paths):
"""Generates the index file for the specified folder"""
# Determine the namespaces listed here (as subfolders)
# Determine the namespaces listed here (as sub folders)
# and the files (.html files) that we should link to
namespaces = []
files = []
@ -320,10 +320,10 @@ def generate_documentation(scheme_file):
docs.write_text('The following %d methods return this type as a result.' % len(functions))
docs.begin_table(2)
for function in functions:
link = get_create_path_for(function)
for func in functions:
link = get_create_path_for(func)
link = get_relative_path(link, relative_to=filename)
docs.add_row(get_class_name(function), link=link)
docs.add_row(get_class_name(func), link=link)
docs.end_table()
docs.end_body()

1
docs/res/core.html
View File

@ -188,6 +188,7 @@ messages = result.messages</pre>
you're still able to invoke these methods manually.</p>
</div>
</div>
<script>
contentDiv = document.getElementById("contentDiv");
searchDiv = document.getElementById("searchDiv");

2
telethon/crypto/__init__.py
View File

@ -1,4 +1,4 @@
from .aes import AES
from .rsa import RSA, RSAServerKey
from .auth_key import AuthKey
from .factorizator import Factorizator
from .factorization import Factorization

6
telethon/crypto/factorizator.py → telethon/crypto/factorization.py
View File

@ -1,7 +1,7 @@
from random import randint
class Factorizator:
class Factorization:
@staticmethod
def find_small_multiplier_lopatin(what):
"""Finds the small multiplier by using Lopatin's method"""
@ -25,7 +25,7 @@ class Factorizator:
x = c
z = y - x if x < y else x - y
g = Factorizator.gcd(z, what)
g = Factorization.gcd(z, what)
if g != 1:
break
@ -58,5 +58,5 @@ class Factorizator:
@staticmethod
def factorize(pq):
"""Factorizes the given number and returns both the divisor and the number divided by the divisor"""
divisor = Factorizator.find_small_multiplier_lopatin(pq)
divisor = Factorization.find_small_multiplier_lopatin(pq)
return divisor, pq // divisor

2
telethon/errors.py
View File

@ -112,7 +112,7 @@ class RPCError(Exception):
'FILE_PARTS_INVALID': 'The number of file parts is invalid.',
'FILE_PART_(\d+)_MISSING':
'Part {} of the file is missing from storage.',
'MD5_CHECKSUM_INVALID': 'The MD5 checksums do not match.',
'MD5_CHECKSUM_INVALID': 'The MD5 check-sums do not match.',
'PHOTO_INVALID_DIMENSIONS': 'The photo dimensions are invalid.',
'FIELD_NAME_INVALID': 'The field with the name FIELD_NAME is invalid.',
'FIELD_NAME_EMPTY': 'The field with the name FIELD_NAME is missing.',

9
telethon/helpers.py
View File

@ -41,8 +41,8 @@ def calc_msg_key(data):
return sha1(data)[4:20]
def generate_key_data_from_nonces(server_nonce, new_nonce):
"""Generates the key data corresponding to the given nonces"""
def generate_key_data_from_nonce(server_nonce, new_nonce):
"""Generates the key data corresponding to the given nonce"""
hash1 = sha1(bytes(new_nonce + server_nonce))
hash2 = sha1(bytes(server_nonce + new_nonce))
hash3 = sha1(bytes(new_nonce + new_nonce))
@ -68,10 +68,11 @@ def sha256(data):
def get_password_hash(pw, current_salt):
"""Gets the password hash for the two-step verification.
curent_salt should be the byte array provided by invoking GetPasswordRequest()"""
current_salt should be the byte array provided by invoking GetPasswordRequest()"""
# Passwords are encoded as UTF-8
# https://github.com/DrKLO/Telegram/blob/e31388/TMessagesProj/src/main/java/org/telegram/ui/LoginActivity.java#L2003
# At https://github.com/DrKLO/Telegram/blob/e31388
# src/main/java/org/telegram/ui/LoginActivity.java#L2003
data = pw.encode('utf-8')
pw_hash = current_salt + data + current_salt

2
telethon/interactive_telegram_client.py
View File

@ -86,7 +86,7 @@ class InteractiveTelegramClient(TelegramClient):
# Display them so the user can choose
for i, entity in enumerate(entities):
i += 1 # 1-based index for normies
i += 1 # 1-based index
try:
print('{}. {}'.format(i, get_display_name(entity)))
except UnicodeEncodeError:

8
telethon/network/authenticator.py
View File

@ -2,7 +2,7 @@ import os
import time
from .. import helpers as utils
from ..crypto import AES, RSA, AuthKey, Factorizator
from ..crypto import AES, RSA, AuthKey, Factorization
from ..network import MtProtoPlainSender
from ..utils import BinaryReader, BinaryWriter
@ -49,7 +49,7 @@ def do_authentication(transport):
# Step 2 sending: DH Exchange
new_nonce = os.urandom(32)
p, q = Factorizator.factorize(pq)
p, q = Factorization.factorize(pq)
with BinaryWriter() as pq_inner_data_writer:
pq_inner_data_writer.write_int(
0x83c95aec, signed=False) # PQ Inner Data
@ -120,12 +120,12 @@ def do_authentication(transport):
encrypted_answer = reader.tgread_bytes()
# Step 3 sending: Complete DH Exchange
key, iv = utils.generate_key_data_from_nonces(server_nonce, new_nonce)
key, iv = utils.generate_key_data_from_nonce(server_nonce, new_nonce)
plain_text_answer = AES.decrypt_ige(encrypted_answer, key, iv)
g, dh_prime, ga, time_offset = None, None, None, None
with BinaryReader(plain_text_answer) as dh_inner_data_reader:
dh_inner_data_reader.read(20) # hashsum
dh_inner_data_reader.read(20) # hash sum
code = dh_inner_data_reader.read_int(signed=False)
if code != 0xb5890dba:
raise AssertionError('Invalid DH Inner Data code: {}'.format(code))

2
telethon/network/mtproto_plain_sender.py
View File

@ -46,7 +46,7 @@ class MtProtoPlainSender:
# See https://core.telegram.org/mtproto/description#message-identifier-msg-id
ms_time = int(time.time() * 1000)
new_msg_id = (((ms_time // 1000) << 32)
| # "must approximately equal unixtime*2^32"
| # "must approximately equal unix time*2^32"
((ms_time % 1000) << 22)
| # "approximate moment in time the message was created"
random.randint(0, 524288)

4
telethon/network/mtproto_sender.py
View File

@ -282,9 +282,9 @@ class MtProtoSender:
def handle_pong(self, msg_id, sequence, reader, request):
self.logger.debug('Handling pong')
reader.read_int(signed=False) # code
recv_msg_id = reader.read_long(signed=False)
received_msg_id = reader.read_long(signed=False)
if recv_msg_id == request.msg_id:
if received_msg_id == request.msg_id:
self.logger.warning('Pong confirmed a request')
request.confirm_received = True

26
telethon/parser/markdown_parser.py
View File

@ -116,30 +116,30 @@ def parse_message_entities(msg):
del msg[entity.offset]
# Iterate over all the entities but the current
for subentity in [e for e in entities if e is not entity]:
for sub_entity in [e for e in entities if e is not entity]:
# First case, one in one out: so*me_th_in*g.
# In this case, the current entity length is decreased by two,
# and all the subentities offset decreases 1
if (subentity.offset > entity.offset and
subentity.offset + subentity.length <
# and all the sub_entities offset decreases 1
if (sub_entity.offset > entity.offset and
sub_entity.offset + sub_entity.length <
entity.offset + entity.length):
entity.length -= 2
subentity.offset -= 1
sub_entity.offset -= 1
# Second case, both inside: so*me_th*in_g.
# In this case, the current entity length is decreased by one,
# and all the subentities offset and length decrease 1
elif (entity.offset < subentity.offset < entity.offset +
entity.length < subentity.offset + subentity.length):
# and all the sub_entities offset and length decrease 1
elif (entity.offset < sub_entity.offset < entity.offset +
entity.length < sub_entity.offset + sub_entity.length):
entity.length -= 1
subentity.offset -= 1
subentity.length -= 1
sub_entity.offset -= 1
sub_entity.length -= 1
# Third case, both outside: so*me*th_in_g.
# In this case, the current entity is left untouched,
# and all the subentities offset decreases 2
elif subentity.offset > entity.offset + entity.length:
subentity.offset -= 2
# and all the sub_entities offset decreases 2
elif sub_entity.offset > entity.offset + entity.length:
sub_entity.offset -= 2
# Finally, we can join our poor mutilated message back and return
msg = ''.join(msg)

16
telethon/telegram_client.py
View File

@ -17,8 +17,7 @@ from .tl.functions import InitConnectionRequest, InvokeWithLayerRequest
from .tl.functions.account import GetPasswordRequest
from .tl.functions.auth import (CheckPasswordRequest, LogOutRequest,
SendCodeRequest, SignInRequest,
SignUpRequest)
from .tl.functions.auth import ImportBotAuthorizationRequest
SignUpRequest, ImportBotAuthorizationRequest)
from .tl.functions.help import GetConfigRequest
from .tl.functions.messages import (
GetDialogsRequest, GetHistoryRequest, ReadHistoryRequest, SendMediaRequest,
@ -34,7 +33,7 @@ from .tl.types import (
MessageMediaContact, MessageMediaDocument, MessageMediaPhoto,
UserProfilePhotoEmpty)
from .utils import (find_user_or_chat, get_input_peer,
get_appropiate_part_size, get_extension)
get_appropriated_part_size, get_extension)
class TelegramClient:
@ -273,7 +272,7 @@ class TelegramClient:
self.session = None
return True
except:
except (RPCError, ConnectionError):
# Something happened when logging out, restore the state back
self.sender.logging_out = False
return False
@ -282,8 +281,7 @@ class TelegramClient:
def list_sessions():
"""Lists all the sessions of the users who have ever connected
using this client and never logged out"""
return [path.splitext(path.basename(f))[
0] # splitext = split ext (not spli text!)
return [path.splitext(path.basename(f))[0]
for f in listdir('.') if f.endswith('.session')]
# endregion
@ -424,7 +422,7 @@ class TelegramClient:
"""
file_size = path.getsize(file_path)
if not part_size_kb:
part_size_kb = get_appropiate_part_size(file_size)
part_size_kb = get_appropriated_part_size(file_size)
if part_size_kb > 512:
raise ValueError('The part size must be less or equal to 512KB')
@ -534,7 +532,7 @@ class TelegramClient:
add_extension=True,
download_big=True):
"""Downloads the profile photo for an user or a chat (including channels).
Returns False if no photo was providen, or if it was Empty"""
Returns False if no photo was provided, or if it was Empty"""
if (not profile_photo or
isinstance(profile_photo, UserProfilePhotoEmpty) or
@ -693,7 +691,7 @@ class TelegramClient:
if not file_size:
raise ValueError('A part size value must be provided')
else:
part_size_kb = get_appropiate_part_size(file_size)
part_size_kb = get_appropriated_part_size(file_size)
part_size = int(part_size_kb * 1024)
if part_size % 1024 != 0:

4
telethon/tl/session.py
View File

@ -31,7 +31,7 @@ class Session:
try:
os.remove('{}.session'.format(self.session_user_id))
return True
except:
except OSError:
return False
@staticmethod
@ -54,7 +54,7 @@ class Session:
# Refer to mtproto_plain_sender.py for the original method, this is a simple copy
ms_time = int(time.time() * 1000)
new_msg_id = (((ms_time // 1000 + self.time_offset) << 32)
| # "must approximately equal unixtime*2^32"
| # "must approximately equal unix time*2^32"
((ms_time % 1000) << 22)
| # "approximate moment in time the message was created"
random.randint(0, 524288)

4
telethon/utils/tl_utils.py
View File

@ -77,8 +77,8 @@ def find_user_or_chat(peer, users, chats):
return None
def get_appropiate_part_size(file_size):
"""Gets the appropiate part size when uploading or downloading files,
def get_appropriated_part_size(file_size):
"""Gets the appropriated part size when uploading or downloading files,
given an initial file size"""
if file_size <= 1048576: # 1MB
return 32

8
telethon_generator/parser/tl_object.py
View File

@ -42,12 +42,12 @@ class TLObject:
([0-9a-f]+) # The constructor ID is in hexadecimal form
(?:\s # After that, we want to match its arguments (name:type)
\{? # For handling the start of the «{X:Type}» case
{? # For handling the start of the «{X:Type}» case
\w+ # The argument name will always be an alpha-only name
: # Then comes the separator between name:type
[\w\d<>#.?!]+ # The type is slightly more complex, since it's alphanumeric and it can
# also have Vector<type>, flags:# and flags.0?default, plus :!X as type
\}? # For handling the end of the «{X:Type}» case
}? # For handling the end of the «{X:Type}» case
)* # Match 0 or more arguments
\s # Leave a space between the arguments and the equal
=
@ -58,12 +58,12 @@ class TLObject:
# Sub-regex to match the arguments (sadly, it cannot be embedded in the first regex)
args_match = re.findall(r'''
(\{)? # We may or may not capture the opening brace
({)? # We may or may not capture the opening brace
(\w+) # First we capture any alpha name with length 1 or more
: # Which is separated from its type by a colon
([\w\d<>#.?!]+) # The type is slightly more complex, since it's alphanumeric and it can
# also have Vector<type>, flags:# and flags.0?default, plus :!X as type
(\})? # We may or not capture the closing brace
(})? # We may or not capture the closing brace
''', tl, re.IGNORECASE | re.VERBOSE)
# Retrieve the matched arguments

2
telethon_generator/tl_generator.py
View File

@ -250,7 +250,7 @@ class TLGenerator:
'"""File generated by TLObjects\' generator. All changes will be ERASED"""')
builder.writeln()
builder.writeln('from ..tl import types, functions')
builder.writeln('from . import types, functions')
builder.writeln()
# Create a variable to indicate which layer this is

32
telethon_tests/crypto_test.py
View File

@ -1,7 +1,7 @@
import unittest
import telethon.helpers as utils
from telethon.crypto import AES, Factorizator
from telethon.crypto import AES, Factorization
class CryptoTests(unittest.TestCase):
@ -23,11 +23,11 @@ class CryptoTests(unittest.TestCase):
def test_sha1():
string = 'Example string'
hashsum = utils.sha1(string.encode('utf-8'))
hash_sum = utils.sha1(string.encode('utf-8'))
expected = b'\nT\x92|\x8d\x06:)\x99\x04\x8e\xf8j?\xc4\x8e\xd3}m9'
assert hashsum == expected, 'Invalid sha1 hashsum representation (should be {}, but is {})'\
.format(expected, hashsum)
assert hash_sum == expected, 'Invalid sha1 hash_sum representation (should be {}, but is {})'\
.format(expected, hash_sum)
def test_aes_encrypt(self):
value = AES.encrypt_ige(self.plain_text, self.key, self.iv)
@ -67,8 +67,12 @@ class CryptoTests(unittest.TestCase):
msg_key = b'\xba\x1a\xcf\xda\xa8^Cbl\xfa\xb6\x0c:\x9b\xb0\xfc'
key, iv = utils.calc_key(shared_key, msg_key, client=True)
expected_key = b"\xaf\xe3\x84Qm\xe0!\x0c\xd91\xe4\x9a\xa0v_gcx\xa1\xb0\xc9\xbc\x16'v\xcf,\x9dM\xae\xc6\xa5"
expected_iv = b'\xb8Q\xf3\xc5\xa3]\xc6\xdf\x9e\xe0Q\xbd"\x8d\x13\t\x0e\x9a\x9d^8\xa2\xf8\xe7\x00w\xd9\xc1\xa7\xa0\xf7\x0f'
expected_key = b"\xaf\xe3\x84Qm\xe0!\x0c\xd91\xe4\x9a\xa0v_gc" \
b"x\xa1\xb0\xc9\xbc\x16'v\xcf,\x9dM\xae\xc6\xa5"
expected_iv = b'\xb8Q\xf3\xc5\xa3]\xc6\xdf\x9e\xe0Q\xbd"\x8d' \
b'\x13\t\x0e\x9a\x9d^8\xa2\xf8\xe7\x00w\xd9\xc1' \
b'\xa7\xa0\xf7\x0f'
assert key == expected_key, 'Invalid key (expected ("{}"), got ("{}"))'.format(
expected_key, key)
@ -79,8 +83,12 @@ class CryptoTests(unittest.TestCase):
msg_key = b'\x86m\x92i\xcf\x8b\x93\xaa\x86K\x1fi\xd04\x83]'
key, iv = utils.calc_key(shared_key, msg_key, client=False)
expected_key = b'\xdd0X\xb6\x93\x8e\xc9y\xef\x83\xf8\x8cj\xa7h\x03\xe2\xc6\xb16\xc5\xbb\xfc\xe7\xdf\xd6\xb1g\xf7u\xcfk'
expected_iv = b'\xdcL\xc2\x18\x01J"X\x86lb\xb6\xb547\xfd\xe2a4\xb6\xaf}FS\xd7[\xe0N\r\x19\xfb\xbc'
expected_key = b'\xdd0X\xb6\x93\x8e\xc9y\xef\x83\xf8\x8cj' \
b'\xa7h\x03\xe2\xc6\xb16\xc5\xbb\xfc\xe7' \
b'\xdf\xd6\xb1g\xf7u\xcfk'
expected_iv = b'\xdcL\xc2\x18\x01J"X\x86lb\xb6\xb547\xfd' \
b'\xe2a4\xb6\xaf}FS\xd7[\xe0N\r\x19\xfb\xbc'
assert key == expected_key, 'Invalid key (expected ("{}"), got ("{}"))'.format(
expected_key, key)
@ -95,11 +103,11 @@ class CryptoTests(unittest.TestCase):
value, expected)
@staticmethod
def test_generate_key_data_from_nonces():
def test_generate_key_data_from_nonce():
server_nonce = b'I am the server nonce.'
new_nonce = b'I am a new calculated nonce.'
key, iv = utils.generate_key_data_from_nonces(server_nonce, new_nonce)
key, iv = utils.generate_key_data_from_nonce(server_nonce, new_nonce)
expected_key = b'?\xc4\xbd\xdf\rWU\x8a\xf5\x0f+V\xdc\x96up\x1d\xeeG\x00\x81|\x1eg\x8a\x8f{\xf0y\x80\xda\xde'
expected_iv = b'Q\x9dpZ\xb7\xdd\xcb\x82_\xfa\xf4\x90\xecn\x10\x9cD\xd2\x01\x8d\x83\xa0\xa4^\xb8\x91,\x7fI am'
@ -109,9 +117,9 @@ class CryptoTests(unittest.TestCase):
key, expected_iv)
@staticmethod
def test_factorizator():
def test_factorize():
pq = 3118979781119966969
p, q = Factorizator.factorize(pq)
p, q = Factorization.factorize(pq)
assert p == 1719614201, 'Factorized pair did not yield the correct result'
assert q == 1813767169, 'Factorized pair did not yield the correct result'

8
telethon_tests/network_test.py
View File

@ -12,10 +12,10 @@ def run_server_echo_thread(port):
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.bind(('', port))
s.listen(1)
conn, addr = s.accept()
with conn:
data = conn.recv(16)
conn.send(data)
connection, address = s.accept()
with connection:
data = connection.recv(16)
connection.send(data)
server = threading.Thread(target=server_thread)
server.start()

8
try_telethon.py
View File

@ -7,18 +7,18 @@ from telethon.interactive_telegram_client import (InteractiveTelegramClient,
def load_settings(path='api/settings'):
"""Loads the user settings located under `api/`"""
settings = {}
result = {}
with open(path, 'r', encoding='utf-8') as file:
for line in file:
value_pair = line.split('=')
left = value_pair[0].strip()
right = value_pair[1].strip()
if right.isnumeric():
settings[left] = int(right)
result[left] = int(right)
else:
settings[left] = right
result[left] = right
return settings
return result
if __name__ == '__main__':