hoelee/Telethon
1
0
Fork 0
mirror of https://github.com/LonamiWebs/Telethon.git synced 2026-09-07 01:11:56 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

First attempt at TelegramClient. Added fixes and doc

Browse Source
This commit is contained in:
Lonami
2016-09-04 11:07:18 +02:00
parent c863537b7b
commit 39a23559f0
19 changed files with 256 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
utils/aes.py
-61
View File
@@ -1,61 +0,0 @@
import pyaes
class AES:
@staticmethod
def decrypt_ige(cipher_text, key, iv):
iv1 = iv[:len(iv)//2]
iv2 = iv[len(iv)//2:]
aes = pyaes.AES(key)
plain_text = [0] * len(cipher_text)
blocks_count = len(cipher_text) // 16
cipher_text_block = [0] * 16
for block_index in range(blocks_count):
for i in range(16):
cipher_text_block[i] = cipher_text[block_index * 16 + i] ^ iv2[i]
plain_text_block = aes.decrypt(cipher_text_block)
for i in range(16):
plain_text_block[i] ^= iv1[i]
iv1 = cipher_text[block_index * 16:block_index * 16 + 16]
iv2 = plain_text_block[0:16]
plain_text[block_index * 16:block_index * 16 + 16] = plain_text_block[:16]
return bytes(plain_text)
@staticmethod
def encrypt_ige(plain_text, key, iv):
# TODO: Random padding
padding = bytes(16 - len(plain_text) % 16)
plain_text += padding
iv1 = iv[:len(iv)//2]
iv2 = iv[len(iv)//2:]
aes = pyaes.AES(key)
blocks_count = len(plain_text) // 16
cipher_text = [0] * len(plain_text)
for block_index in range(blocks_count):
plain_text_block = list(plain_text[block_index * 16:block_index * 16 + 16])
for i in range(16):
plain_text_block[i] ^= iv1[i]
cipher_text_block = aes.encrypt(plain_text_block)
for i in range(16):
cipher_text_block[i] ^= iv2[i]
iv1 = cipher_text_block[0:16]
iv2 = plain_text[block_index * 16:block_index * 16 + 16]
cipher_text[block_index * 16:block_index * 16 + 16] = cipher_text_block[:16]
return bytes(cipher_text)
utils/auth_key.py
-29
View File
@@ -1,29 +0,0 @@
# This file is based on TLSharp
# https://github.com/sochix/TLSharp/blob/master/TLSharp.Core/MTProto/Crypto/AuthKey.cs
import utils.helpers as utils
from utils.binary_writer import BinaryWriter
from utils.binary_reader import BinaryReader
class AuthKey:
def __init__(self, gab=None, data=None):
if gab:
self.key = utils.get_byte_array(gab, signed=False)
elif data:
self.key = data
else:
raise AssertionError('Either a gab integer or data bytes array must be provided')
with BinaryReader(utils.sha1(self.key)) as reader:
self.aux_hash = reader.read_long(signed=False)
reader.read(4)
self.key_id = reader.read_long(signed=False)
def calc_new_nonce_hash(self, new_nonce, number):
with BinaryWriter() as writer:
writer.write(new_nonce)
writer.write_byte(number)
writer.write_long(self.aux_hash, signed=False)
new_nonce_hash = utils.sha1(writer.get_bytes())[4:20]
return new_nonce_hash
utils/binary_reader.py
+12 -6
View File
@@ -1,6 +1,7 @@
from io import BytesIO, BufferedReader
from tl.all_tlobjects import tlobjects
from struct import unpack
import inspect
import os
@@ -85,14 +86,20 @@ class BinaryReader:
def tgread_object(self):
"""Reads a Telegram object"""
id = self.read_int()
clazz = tlobjects.get(id, None)
constructor_id = self.read_int()
clazz = tlobjects.get(constructor_id, None)
if clazz is None:
raise ImportError('Could not find a matching ID for the TLObject that was supposed to be read. '
'Found ID: {}'.format(hex(id)))
'Found ID: {}'.format(hex(constructor_id)))
# Instantiate the class and return the result
result = clazz()
# Now we need to determine the number of parameters of the class, so we can
# instantiate it with all of them set to None, and still, no need to write
# the default =None in all the classes, thus forcing the user to provide a real value
sig = inspect.signature(clazz.__init__)
params = [None] * (len(sig.parameters) - 1) # Subtract 1 (self)
result = clazz(*params) # https://docs.python.org/3/tutorial/controlflow.html#unpacking-argument-lists
# Finally, read the object and return the result
result.on_response(self)
return result
@@ -100,7 +107,6 @@ class BinaryReader:
def close(self):
self.reader.close()
# TODO Do I need to close the underlying stream?
# region Position related
utils/binary_writer.py
+2 -10
View File
@@ -65,18 +65,11 @@ class BinaryWriter:
if padding != 0:
padding = 4 - padding
# TODO ensure that _this_ is right (it appears to be)
self.write(bytes([254]))
self.write(bytes([len(data) % 256]))
self.write(bytes([(len(data) >> 8) % 256]))
self.write(bytes([(len(data) >> 16) % 256]))
self.write(data)
""" Original:
binaryWriter.Write((byte)254);
binaryWriter.Write((byte)(bytes.Length));
binaryWriter.Write((byte)(bytes.Length >> 8));
binaryWriter.Write((byte)(bytes.Length >> 16));
"""
self.write(bytes(padding))
@@ -84,10 +77,10 @@ class BinaryWriter:
"""Write a string by using Telegram guidelines"""
return self.tgwrite_bytes(string.encode('utf-8'))
def tgwrite_bool(self, bool):
def tgwrite_bool(self, boolean):
"""Write a boolean value by using Telegram guidelines"""
# boolTrue boolFalse
return self.write_int(0x997275b5 if bool else 0xbc799737, signed=False)
return self.write_int(0x997275b5 if boolean else 0xbc799737, signed=False)
# endregion
@@ -98,7 +91,6 @@ class BinaryWriter:
def close(self):
"""Close the current stream"""
self.writer.close()
# TODO Do I need to close the underlying stream?
def get_bytes(self, flush=True):
"""Get the current bytes array content from the buffer, optionally flushing first"""
utils/factorizator.py
-61
View File
@@ -1,61 +0,0 @@
# This file is based on TLSharp
# https://github.com/sochix/TLSharp/blob/master/TLSharp.Core/MTProto/Crypto/Factorizator.cs
from random import randint
class Factorizator:
@staticmethod
def find_small_multiplier_lopatin(what):
g = 0
for i in range(3):
q = (randint(0, 127) & 15) + 17
x = randint(0, 1000000000) + 1
y = x
lim = 1 << (i + 18)
for j in range(1, lim):
a, b, c = x, x, q
while b != 0:
if (b & 1) != 0:
c += a
if c >= what:
c -= what
a += a
if a >= what:
a -= what
b >>= 1
x = c
z = y - x if x < y else x - y
g = Factorizator.gcd(z, what)
if g != 1:
break
if (j & (j - 1)) == 0:
y = x
if g > 1:
break
p = what // g
return min(p, g)
@staticmethod
def gcd(a, b):
while a != 0 and b != 0:
while b & 1 == 0:
b >>= 1
while a & 1 == 0:
a >>= 1
if a > b:
a -= b
else:
b -= a
return a if b == 0 else b
@staticmethod
def factorize(pq):
divisor = Factorizator.find_small_multiplier_lopatin(pq)
return divisor, pq // divisor
utils/helpers.py
+19
View File
@@ -17,6 +17,7 @@ def generate_random_bytes(count):
def get_byte_array(integer, signed):
"""Gets the arbitrary-length byte array corresponding to the given integer"""
bits = integer.bit_length()
byte_length = (bits + bits_per_byte - 1) // bits_per_byte
# For some strange reason, this has to be big!
@@ -49,6 +50,7 @@ def calc_msg_key_offset(data, offset, limit):
def generate_key_data_from_nonces(server_nonce, new_nonce):
"""Generates the key data corresponding to the given nonces"""
hash1 = sha1(bytes(new_nonce + server_nonce))
hash2 = sha1(bytes(server_nonce + new_nonce))
hash3 = sha1(bytes(new_nonce + new_nonce))
@@ -66,6 +68,23 @@ def generate_key_data_from_nonces(server_nonce, new_nonce):
def sha1(data):
"""Calculates the SHA1 digest for the given data"""
sha = hashlib.sha1()
sha.update(data)
return sha.digest()
def load_settings():
"""Loads the user settings located under `api/`"""
settings = {}
with open('api/settings', '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)
else:
settings[left] = right
return settings
utils/rsa.py
-62
View File
@@ -1,62 +0,0 @@
# This file is based on TLSharp
# https://github.com/sochix/TLSharp/blob/master/TLSharp.Core/Auth/Authenticator.cs
from utils.binary_writer import BinaryWriter
import utils.helpers as utils
class RSAServerKey:
def __init__(self, fingerprint, m, e):
self.fingerprint = fingerprint
self.m = m
self.e = e
def encrypt(self, data, offset=None, length=None):
if offset is None:
offset = 0
if length is None:
length = len(data)
with BinaryWriter() as writer:
# Write SHA
writer.write(utils.sha1(data[offset:offset+length]))
# Write data
writer.write(data[offset:offset+length])
# Add padding if required
if length < 235:
writer.write(utils.generate_random_bytes(235 - length))
cipher_text = utils.get_byte_array(
pow(int.from_bytes(writer.get_bytes(), byteorder='big'), self.e, self.m),
signed=False)
if len(cipher_text) == 256:
return cipher_text
else:
padding = bytes(256 - len(cipher_text))
return padding + cipher_text
class RSA:
_server_keys = {
'216be86c022bb4c3':
RSAServerKey('216be86c022bb4c3', int('C150023E2F70DB7985DED064759CFECF0AF328E69A41DAF4D6F01B538135A6F9'
'1F8F8B2A0EC9BA9720CE352EFCF6C5680FFC424BD634864902DE0B4BD6D49F4E'
'580230E3AE97D95C8B19442B3C0A10D8F5633FECEDD6926A7F6DAB0DDB7D457F'
'9EA81B8465FCD6FFFEED114011DF91C059CAEDAF97625F6C96ECC74725556934'
'EF781D866B34F011FCE4D835A090196E9A5F0E4449AF7EB697DDB9076494CA5F'
'81104A305B6DD27665722C46B60E5DF680FB16B210607EF217652E60236C255F'
'6A28315F4083A96791D7214BF64C1DF4FD0DB1944FB26A2A57031B32EEE64AD1'
'5A8BA68885CDE74A5BFC920F6ABF59BA5C75506373E7130F9042DA922179251F',
16), int('010001', 16))
}
@staticmethod
def encrypt(fingerprint, data, offset=None, length=None):
if fingerprint.lower() not in RSA._server_keys:
return None
key = RSA._server_keys[fingerprint.lower()]
return key.encrypt(data, offset, length)