mirror of
https://github.com/LonamiWebs/Telethon.git
synced 2025-08-08 12:59:46 +00:00
Added and updated documentation
This commit is contained in:
@@ -21,18 +21,23 @@ class BinaryReader:
|
||||
# region Reading
|
||||
|
||||
def read_int(self, signed=True):
|
||||
"""Reads an integer (4 bytes) value"""
|
||||
return int.from_bytes(self.reader.read(4), signed=signed, byteorder='big')
|
||||
|
||||
def read_long(self, signed=True):
|
||||
"""Reads a long integer (8 bytes) value"""
|
||||
return int.from_bytes(self.reader.read(8), signed=signed, byteorder='big')
|
||||
|
||||
def read_large_int(self, bits):
|
||||
"""Reads a n-bits long integer value"""
|
||||
return int.from_bytes(self.reader.read(bits // 8), byteorder='big')
|
||||
|
||||
def read(self, length):
|
||||
"""Read the given amount of bytes"""
|
||||
return self.reader.read(length)
|
||||
|
||||
def get_bytes(self):
|
||||
"""Gets the byte array representing the current buffer as a whole"""
|
||||
return self.stream.getbuffer()
|
||||
|
||||
# endregion
|
||||
@@ -40,6 +45,7 @@ class BinaryReader:
|
||||
# region Telegram custom reading
|
||||
|
||||
def tgread_bytes(self):
|
||||
"""Reads a Telegram-encoded byte array, without the need of specifying its length"""
|
||||
first_byte = self.read(1)
|
||||
if first_byte == 254:
|
||||
length = self.read(1) | (self.read(1) << 8) | (self.read(1) << 16)
|
||||
@@ -56,6 +62,7 @@ class BinaryReader:
|
||||
return data
|
||||
|
||||
def tgread_string(self):
|
||||
"""Reads a Telegram-encoded string"""
|
||||
return str(self.tgread_bytes(), encoding='utf-8')
|
||||
|
||||
def tgread_object(self):
|
||||
|
@@ -18,9 +18,11 @@ class BinaryWriter:
|
||||
# region Writing
|
||||
|
||||
def write_byte(self, value):
|
||||
"""Writes a single byte value"""
|
||||
self.writer.write(pack('B', value))
|
||||
|
||||
def write_int(self, value, signed=True):
|
||||
"""Writes an integer value (4 bytes), which can or cannot be signed"""
|
||||
if signed:
|
||||
self.writer.write(pack('i', value))
|
||||
else:
|
||||
@@ -28,6 +30,7 @@ class BinaryWriter:
|
||||
self.writer.write(pack('I', value))
|
||||
|
||||
def write_long(self, value, signed=True):
|
||||
"""Writes a long integer value (8 bytes), which can or cannot be signed"""
|
||||
if signed:
|
||||
self.writer.write(pack('q', value))
|
||||
else:
|
||||
@@ -35,15 +38,19 @@ class BinaryWriter:
|
||||
self.writer.write(pack('Q', value))
|
||||
|
||||
def write_float(self, value):
|
||||
"""Writes a floating point value (4 bytes)"""
|
||||
self.writer.write(pack('f', value))
|
||||
|
||||
def write_double(self, value):
|
||||
"""Writes a floating point value (8 bytes)"""
|
||||
self.writer.write(pack('d', value))
|
||||
|
||||
def write_large_int(self, value, bits):
|
||||
"""Writes a n-bits long integer value"""
|
||||
self.writer.write(pack('{}B'.format(bits // 8), value))
|
||||
|
||||
def write(self, data):
|
||||
"""Writes the given bytes array"""
|
||||
self.writer.write(data)
|
||||
|
||||
# endregion
|
||||
@@ -51,7 +58,7 @@ class BinaryWriter:
|
||||
# region Telegram custom writing
|
||||
|
||||
def tgwrite_bytes(self, data):
|
||||
|
||||
"""Write bytes by using Telegram guidelines"""
|
||||
if len(data) < 254:
|
||||
padding = (len(data) + 1) % 4
|
||||
if padding != 0:
|
||||
@@ -71,7 +78,6 @@ class BinaryWriter:
|
||||
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));
|
||||
@@ -82,22 +88,27 @@ class BinaryWriter:
|
||||
self.write(bytes(padding))
|
||||
|
||||
def tgwrite_string(self, string):
|
||||
"""Write a string by using Telegram guidelines"""
|
||||
return self.tgwrite_bytes(string.encode('utf-8'))
|
||||
|
||||
def tgwrite_bool(self, bool):
|
||||
"""Write a boolean value by using Telegram guidelines"""
|
||||
# boolTrue boolFalse
|
||||
return self.write_int(0x997275b5 if bool else 0xbc799737, signed=False)
|
||||
|
||||
# endregion
|
||||
|
||||
def flush(self):
|
||||
"""Flush the current stream to "update" changes"""
|
||||
self.writer.flush()
|
||||
|
||||
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"""
|
||||
if flush:
|
||||
self.writer.flush()
|
||||
self.stream.getbuffer()
|
||||
|
@@ -4,6 +4,7 @@ from hashlib import sha1
|
||||
|
||||
|
||||
def generate_random_long(signed=True):
|
||||
"""Generates a random long integer (8 bytes), which is optionally signed"""
|
||||
result = random.getrandbits(64)
|
||||
if not signed:
|
||||
result &= 0xFFFFFFFFFFFFFFFF # Ensure it's unsigned
|
||||
@@ -12,6 +13,7 @@ def generate_random_long(signed=True):
|
||||
|
||||
|
||||
def generate_random_bytes(count):
|
||||
"""Generates a random bytes array"""
|
||||
with BinaryWriter() as writer:
|
||||
for _ in range(count):
|
||||
writer.write(random.getrandbits(8))
|
||||
@@ -20,6 +22,7 @@ def generate_random_bytes(count):
|
||||
|
||||
|
||||
def calc_key(shared_key, msg_key, client):
|
||||
"""Calculate the key based on Telegram guidelines, specifying whether it's the client or not"""
|
||||
x = 0 if client else 8
|
||||
|
||||
buffer = [0] * 48
|
||||
@@ -47,10 +50,12 @@ def calc_key(shared_key, msg_key, client):
|
||||
|
||||
|
||||
def calc_msg_key(data):
|
||||
"""Calculates the message key from the given data"""
|
||||
return sha1(data)[4:20]
|
||||
|
||||
|
||||
def calc_msg_key_offset(data, offset, limit):
|
||||
"""Calculates the message key from offset given data, with an optional offset and limit"""
|
||||
# TODO untested, may not be offset like this
|
||||
# In the original code it was as parameters for the sha function, not slicing the array
|
||||
return sha1(data[offset:offset + limit])[4:20]
|
||||
|
Reference in New Issue
Block a user