mirror of
https://github.com/LonamiWebs/Telethon.git
synced 2025-11-04 01:47:27 +03:00
213 lines
7.7 KiB
Python
213 lines
7.7 KiB
Python
"""
|
|
This module contains several functions that authenticate the client machine
|
|
with Telegram's servers, effectively creating an authorization key.
|
|
"""
|
|
import os
|
|
import time
|
|
from hashlib import sha1
|
|
|
|
from ..tl.types import (
|
|
ResPQ, PQInnerData, ServerDHParamsFail, ServerDHParamsOk,
|
|
ServerDHInnerData, ClientDHInnerData, DhGenOk, DhGenRetry, DhGenFail
|
|
)
|
|
from .. import helpers
|
|
from ..crypto import AES, AuthKey, Factorization, rsa
|
|
from ..errors import SecurityError
|
|
from ..extensions import BinaryReader
|
|
from ..tl.functions import (
|
|
ReqPqMultiRequest, ReqDHParamsRequest, SetClientDHParamsRequest
|
|
)
|
|
|
|
|
|
async def do_authentication(sender):
|
|
"""
|
|
Executes the authentication process with the Telegram servers.
|
|
|
|
:param sender: a connected `MTProtoPlainSender`.
|
|
:return: returns a (authorization key, time offset) tuple.
|
|
"""
|
|
# Step 1 sending: PQ Request, endianness doesn't matter since it's random
|
|
nonce = int.from_bytes(os.urandom(16), 'big', signed=True)
|
|
res_pq = await sender.send(ReqPqMultiRequest(nonce))
|
|
assert isinstance(res_pq, ResPQ), 'Step 1 answer was %s' % res_pq
|
|
|
|
if res_pq.nonce != nonce:
|
|
raise SecurityError('Step 1 invalid nonce from server')
|
|
|
|
pq = get_int(res_pq.pq)
|
|
|
|
# Step 2 sending: DH Exchange
|
|
p, q = Factorization.factorize(pq)
|
|
p, q = rsa.get_byte_array(p), rsa.get_byte_array(q)
|
|
new_nonce = int.from_bytes(os.urandom(32), 'little', signed=True)
|
|
|
|
pq_inner_data = bytes(PQInnerData(
|
|
pq=rsa.get_byte_array(pq), p=p, q=q,
|
|
nonce=res_pq.nonce,
|
|
server_nonce=res_pq.server_nonce,
|
|
new_nonce=new_nonce
|
|
))
|
|
|
|
# sha_digest + data + random_bytes
|
|
cipher_text, target_fingerprint = None, None
|
|
for fingerprint in res_pq.server_public_key_fingerprints:
|
|
cipher_text = rsa.encrypt(fingerprint, pq_inner_data)
|
|
if cipher_text is not None:
|
|
target_fingerprint = fingerprint
|
|
break
|
|
|
|
if cipher_text is None:
|
|
# Second attempt, but now we're allowed to use old keys
|
|
for fingerprint in res_pq.server_public_key_fingerprints:
|
|
cipher_text = rsa.encrypt(fingerprint, pq_inner_data, use_old=True)
|
|
if cipher_text is not None:
|
|
target_fingerprint = fingerprint
|
|
break
|
|
|
|
if cipher_text is None:
|
|
raise SecurityError(
|
|
'Step 2 could not find a valid key for fingerprints: {}'
|
|
.format(', '.join(
|
|
[str(f) for f in res_pq.server_public_key_fingerprints])
|
|
)
|
|
)
|
|
|
|
server_dh_params = await sender.send(ReqDHParamsRequest(
|
|
nonce=res_pq.nonce,
|
|
server_nonce=res_pq.server_nonce,
|
|
p=p, q=q,
|
|
public_key_fingerprint=target_fingerprint,
|
|
encrypted_data=cipher_text
|
|
))
|
|
|
|
assert isinstance(
|
|
server_dh_params, (ServerDHParamsOk, ServerDHParamsFail)),\
|
|
'Step 2.1 answer was %s' % server_dh_params
|
|
|
|
if server_dh_params.nonce != res_pq.nonce:
|
|
raise SecurityError('Step 2 invalid nonce from server')
|
|
|
|
if server_dh_params.server_nonce != res_pq.server_nonce:
|
|
raise SecurityError('Step 2 invalid server nonce from server')
|
|
|
|
if isinstance(server_dh_params, ServerDHParamsFail):
|
|
nnh = int.from_bytes(
|
|
sha1(new_nonce.to_bytes(32, 'little', signed=True)).digest()[4:20],
|
|
'little', signed=True
|
|
)
|
|
if server_dh_params.new_nonce_hash != nnh:
|
|
raise SecurityError('Step 2 invalid DH fail nonce from server')
|
|
|
|
assert isinstance(server_dh_params, ServerDHParamsOk),\
|
|
'Step 2.2 answer was %s' % server_dh_params
|
|
|
|
# Step 3 sending: Complete DH Exchange
|
|
key, iv = helpers.generate_key_data_from_nonce(
|
|
res_pq.server_nonce, new_nonce
|
|
)
|
|
if len(server_dh_params.encrypted_answer) % 16 != 0:
|
|
# See PR#453
|
|
raise SecurityError('Step 3 AES block size mismatch')
|
|
|
|
plain_text_answer = AES.decrypt_ige(
|
|
server_dh_params.encrypted_answer, key, iv
|
|
)
|
|
|
|
with BinaryReader(plain_text_answer) as reader:
|
|
reader.read(20) # hash sum
|
|
server_dh_inner = reader.tgread_object()
|
|
assert isinstance(server_dh_inner, ServerDHInnerData),\
|
|
'Step 3 answer was %s' % server_dh_inner
|
|
|
|
if server_dh_inner.nonce != res_pq.nonce:
|
|
raise SecurityError('Step 3 Invalid nonce in encrypted answer')
|
|
|
|
if server_dh_inner.server_nonce != res_pq.server_nonce:
|
|
raise SecurityError('Step 3 Invalid server nonce in encrypted answer')
|
|
|
|
dh_prime = get_int(server_dh_inner.dh_prime, signed=False)
|
|
g = server_dh_inner.g
|
|
g_a = get_int(server_dh_inner.g_a, signed=False)
|
|
time_offset = server_dh_inner.server_time - int(time.time())
|
|
|
|
b = get_int(os.urandom(256), signed=False)
|
|
g_b = pow(g, b, dh_prime)
|
|
gab = pow(g_a, b, dh_prime)
|
|
|
|
# IMPORTANT: Apart from the conditions on the Diffie-Hellman prime
|
|
# dh_prime and generator g, both sides are to check that g, g_a and
|
|
# g_b are greater than 1 and less than dh_prime - 1. We recommend
|
|
# checking that g_a and g_b are between 2^{2048-64} and
|
|
# dh_prime - 2^{2048-64} as well.
|
|
# (https://core.telegram.org/mtproto/auth_key#dh-key-exchange-complete)
|
|
if not (1 < g < (dh_prime - 1)):
|
|
raise SecurityError('g_a is not within (1, dh_prime - 1)')
|
|
|
|
if not (1 < g_a < (dh_prime - 1)):
|
|
raise SecurityError('g_a is not within (1, dh_prime - 1)')
|
|
|
|
if not (1 < g_b < (dh_prime - 1)):
|
|
raise SecurityError('g_b is not within (1, dh_prime - 1)')
|
|
|
|
safety_range = 2 ** (2048 - 64)
|
|
if not (safety_range <= g_a <= (dh_prime - safety_range)):
|
|
raise SecurityError('g_a is not within (2^{2048-64}, dh_prime - 2^{2048-64})')
|
|
|
|
if not (safety_range <= g_b <= (dh_prime - safety_range)):
|
|
raise SecurityError('g_b is not within (2^{2048-64}, dh_prime - 2^{2048-64})')
|
|
|
|
# Prepare client DH Inner Data
|
|
client_dh_inner = bytes(ClientDHInnerData(
|
|
nonce=res_pq.nonce,
|
|
server_nonce=res_pq.server_nonce,
|
|
retry_id=0, # TODO Actual retry ID
|
|
g_b=rsa.get_byte_array(g_b)
|
|
))
|
|
|
|
client_dh_inner_hashed = sha1(client_dh_inner).digest() + client_dh_inner
|
|
|
|
# Encryption
|
|
client_dh_encrypted = AES.encrypt_ige(client_dh_inner_hashed, key, iv)
|
|
|
|
# Prepare Set client DH params
|
|
dh_gen = await sender.send(SetClientDHParamsRequest(
|
|
nonce=res_pq.nonce,
|
|
server_nonce=res_pq.server_nonce,
|
|
encrypted_data=client_dh_encrypted,
|
|
))
|
|
|
|
nonce_types = (DhGenOk, DhGenRetry, DhGenFail)
|
|
assert isinstance(dh_gen, nonce_types), 'Step 3.1 answer was %s' % dh_gen
|
|
name = dh_gen.__class__.__name__
|
|
if dh_gen.nonce != res_pq.nonce:
|
|
raise SecurityError('Step 3 invalid {} nonce from server'.format(name))
|
|
|
|
if dh_gen.server_nonce != res_pq.server_nonce:
|
|
raise SecurityError(
|
|
'Step 3 invalid {} server nonce from server'.format(name))
|
|
|
|
auth_key = AuthKey(rsa.get_byte_array(gab))
|
|
nonce_number = 1 + nonce_types.index(type(dh_gen))
|
|
new_nonce_hash = auth_key.calc_new_nonce_hash(new_nonce, nonce_number)
|
|
|
|
dh_hash = getattr(dh_gen, 'new_nonce_hash{}'.format(nonce_number))
|
|
if dh_hash != new_nonce_hash:
|
|
raise SecurityError('Step 3 invalid new nonce hash')
|
|
|
|
if not isinstance(dh_gen, DhGenOk):
|
|
raise AssertionError('Step 3.2 answer was %s' % dh_gen)
|
|
|
|
return auth_key, time_offset
|
|
|
|
|
|
def get_int(byte_array, signed=True):
|
|
"""
|
|
Gets the specified integer from its byte array.
|
|
This should be used by this module alone, as it works with big endian.
|
|
|
|
:param byte_array: the byte array representing th integer.
|
|
:param signed: whether the number is signed or not.
|
|
:return: the integer representing the given byte array.
|
|
"""
|
|
return int.from_bytes(byte_array, byteorder='big', signed=signed)
|