Telethon/telethon/network/connection/connection.py

import abc
import asyncio
import socket
import sys

try:
    import ssl as ssl_mod
except ImportError:
    ssl_mod = None

try:
    import python_socks
except ImportError:
    python_socks = None

from ...errors import InvalidChecksumError, InvalidBufferError
from ... import helpers


class Connection(abc.ABC):
    """
    The `Connection` class is a wrapper around ``asyncio.open_connection``.

    Subclasses will implement different transport modes as atomic operations,
    which this class eases doing since the exposed interface simply puts and
    gets complete data payloads to and from queues.

    The only error that will raise from send and receive methods is
    ``ConnectionError``, which will raise when attempting to send if
    the client is disconnected (includes remote disconnections).
    """
    # this static attribute should be redefined by `Connection` subclasses and
    # should be one of `PacketCodec` implementations
    packet_codec = None

    def __init__(self, ip, port, dc_id, *, loggers, proxy=None, local_addr=None):
        self._ip = ip
        self._port = port
        self._dc_id = dc_id  # only for MTProxy, it's an abstraction leak
        self._log = loggers[__name__]
        self._proxy = proxy
        self._local_addr = local_addr
        self._reader = None
        self._writer = None
        self._connected = False
        self._send_task = None
        self._recv_task = None
        self._codec = None
        self._obfuscation = None  # TcpObfuscated and MTProxy
        self._send_queue = asyncio.Queue(1)
        self._recv_queue = asyncio.Queue(1)

    @staticmethod
    def _wrap_socket_ssl(sock):
        if ssl_mod is None:
            raise RuntimeError(
                'Cannot use proxy that requires SSL '
                'without the SSL module being available'
            )

        return ssl_mod.wrap_socket(
            sock,
            do_handshake_on_connect=True,
            ssl_version=ssl_mod.PROTOCOL_SSLv23,
            ciphers='ADH-AES256-SHA')

    @staticmethod
    def _parse_proxy(proxy_type, addr, port, rdns=True, username=None, password=None):
        if isinstance(proxy_type, str):
            proxy_type = proxy_type.lower()

        # Always prefer `python_socks` when available
        if python_socks:
            from python_socks import ProxyType

            # We do the check for numerical values here
            # to be backwards compatible with PySocks proxy format,
            # (since socks.SOCKS5 == 2, socks.SOCKS4 == 1, socks.HTTP == 3)
            if proxy_type == ProxyType.SOCKS5 or proxy_type == 2 or proxy_type == "socks5":
                protocol = ProxyType.SOCKS5
            elif proxy_type == ProxyType.SOCKS4 or proxy_type == 1 or proxy_type == "socks4":
                protocol = ProxyType.SOCKS4
            elif proxy_type == ProxyType.HTTP or proxy_type == 3 or proxy_type == "http":
                protocol = ProxyType.HTTP
            else:
                raise ValueError("Unknown proxy protocol type: {}".format(proxy_type))

            # This tuple must be compatible with `python_socks`' `Proxy.create()` signature
            return protocol, addr, port, username, password, rdns

        else:
            from socks import SOCKS5, SOCKS4, HTTP

            if proxy_type == 2 or proxy_type == "socks5":
                protocol = SOCKS5
            elif proxy_type == 1 or proxy_type == "socks4":
                protocol = SOCKS4
            elif proxy_type == 3 or proxy_type == "http":
                protocol = HTTP
            else:
                raise ValueError("Unknown proxy protocol type: {}".format(proxy_type))

            # This tuple must be compatible with `PySocks`' `socksocket.set_proxy()` signature
            return protocol, addr, port, rdns, username, password

    async def _proxy_connect(self, timeout=None, local_addr=None):
        if isinstance(self._proxy, (tuple, list)):
            parsed = self._parse_proxy(*self._proxy)
        elif isinstance(self._proxy, dict):
            parsed = self._parse_proxy(**self._proxy)
        else:
            raise TypeError("Proxy of unknown format: {}".format(type(self._proxy)))

        # Always prefer `python_socks` when available
        if python_socks:
            # python_socks internal errors are not inherited from
            # builtin IOError (just from Exception). Instead of adding those
            # in exceptions clauses everywhere through the code, we
            # rather monkey-patch them in place. Keep in mind that
            # ProxyError takes error_code as keyword argument.

            class ConnectionErrorExtra(ConnectionError):
                def __init__(self, message, error_code=None):
                    super().__init__(message)
                    self.error_code = error_code

            python_socks._errors.ProxyError = ConnectionErrorExtra
            python_socks._errors.ProxyConnectionError = ConnectionError
            python_socks._errors.ProxyTimeoutError = ConnectionError

            from python_socks.async_.asyncio import Proxy

            proxy = Proxy.create(*parsed)

            # WARNING: If `local_addr` is set we use manual socket creation, because,
            # unfortunately, `Proxy.connect()` does not expose `local_addr`
            # argument, so if we want to bind socket locally, we need to manually
            # create, bind and connect socket, and then pass to `Proxy.connect()` method.

            if local_addr is None:
                sock = await proxy.connect(
                    dest_host=self._ip,
                    dest_port=self._port,
                    timeout=timeout
                )
            else:
                # Here we start manual setup of the socket.
                # The `address` represents the proxy ip and proxy port,
                # not the destination one (!), because the socket
                # connects to the proxy server, not destination server.
                # IPv family is also checked on proxy address.
                if ':' in proxy.proxy_host:
                    mode, address = socket.AF_INET6, (proxy.proxy_host, proxy.proxy_port, 0, 0)
                else:
                    mode, address = socket.AF_INET, (proxy.proxy_host, proxy.proxy_port)

                # Create a non-blocking socket and bind it (if local address is specified).
                sock = socket.socket(mode, socket.SOCK_STREAM)
                sock.setblocking(False)
                sock.bind(local_addr)

                # Actual TCP connection is performed here.
                await asyncio.wait_for(
                    helpers.get_running_loop().sock_connect(sock=sock, address=address),
                    timeout=timeout
                )

                # As our socket is already created and connected,
                # this call sets the destination host/port and
                # starts protocol negotiations with the proxy server.
                sock = await proxy.connect(
                    dest_host=self._ip,
                    dest_port=self._port,
                    timeout=timeout,
                    _socket=sock
                )

        else:
            import socks

            # Here `address` represents destination address (not proxy), because of
            # the `PySocks` implementation of the connection routine.
            # IPv family is checked on proxy address, not destination address.
            if ':' in parsed[1]:
                mode, address = socket.AF_INET6, (self._ip, self._port, 0, 0)
            else:
                mode, address = socket.AF_INET, (self._ip, self._port)

            # Setup socket, proxy, timeout and bind it (if necessary).
            sock = socks.socksocket(mode, socket.SOCK_STREAM)
            sock.set_proxy(*parsed)
            sock.settimeout(timeout)

            if local_addr is not None:
                sock.bind(local_addr)

            # Actual TCP connection and negotiation performed here.
            await asyncio.wait_for(
                helpers.get_running_loop().sock_connect(sock=sock, address=address),
                timeout=timeout
            )

            sock.setblocking(False)

        return sock

    async def _connect(self, timeout=None, ssl=None):
        if self._local_addr is not None:
            # NOTE: If port is not specified, we use 0 port
            # to notify the OS that port should be chosen randomly
            # from the available ones.
            if isinstance(self._local_addr, tuple) and len(self._local_addr) == 2:
                local_addr = self._local_addr
            elif isinstance(self._local_addr, str):
                local_addr = (self._local_addr, 0)
            else:
                raise ValueError("Unknown local address format: {}".format(self._local_addr))
        else:
            local_addr = None

        if not self._proxy:
            self._reader, self._writer = await asyncio.wait_for(
                asyncio.open_connection(
                    host=self._ip,
                    port=self._port,
                    ssl=ssl,
                    local_addr=local_addr
                ), timeout=timeout)
        else:
            # Proxy setup, connection and negotiation is performed here.
            sock = await self._proxy_connect(
                timeout=timeout,
                local_addr=local_addr
            )

            # Wrap socket in SSL context (if provided)
            if ssl:
                sock = self._wrap_socket_ssl(sock)

            self._reader, self._writer = await asyncio.open_connection(sock=sock)

        self._codec = self.packet_codec(self)
        self._init_conn()
        await self._writer.drain()

    async def connect(self, timeout=None, ssl=None):
        """
        Establishes a connection with the server.
        """
        await self._connect(timeout=timeout, ssl=ssl)
        self._connected = True

        loop = helpers.get_running_loop()
        self._send_task = loop.create_task(self._send_loop())
        self._recv_task = loop.create_task(self._recv_loop())

    async def disconnect(self):
        """
        Disconnects from the server, and clears
        pending outgoing and incoming messages.
        """
        if not self._connected:
            return

        self._connected = False

        await helpers._cancel(
            self._log,
            send_task=self._send_task,
            recv_task=self._recv_task
        )

        if self._writer:
            self._writer.close()
            if sys.version_info >= (3, 7):
                try:
                    await asyncio.wait_for(self._writer.wait_closed(), timeout=10)
                except asyncio.TimeoutError:
                    # See issue #3917. For some users, this line was hanging indefinitely.
                    # The hard timeout is not ideal (connection won't be properly closed),
                    # but the code will at least be able to procceed.
                    self._log.warning('Graceful disconnection timed out, forcibly ignoring cleanup')
                except Exception as e:
                    # Disconnecting should never raise. Seen:
                    # * OSError: No route to host and
                    # * OSError: [Errno 32] Broken pipe
                    # * ConnectionResetError
                    self._log.info('%s during disconnect: %s', type(e), e)

    def send(self, data):
        """
        Sends a packet of data through this connection mode.

        This method returns a coroutine.
        """
        if not self._connected:
            raise ConnectionError('Not connected')

        return self._send_queue.put(data)

    async def recv(self):
        """
        Receives a packet of data through this connection mode.

        This method returns a coroutine.
        """
        while self._connected:
            result, err = await self._recv_queue.get()
            if err:
                raise err
            if result:
                return result

        raise ConnectionError('Not connected')

    async def _send_loop(self):
        """
        This loop is constantly popping items off the queue to send them.
        """
        try:
            while self._connected:
                self._send(await self._send_queue.get())
                await self._writer.drain()
        except asyncio.CancelledError:
            pass
        except Exception as e:
            if isinstance(e, IOError):
                self._log.info('The server closed the connection while sending')
            else:
                self._log.exception('Unexpected exception in the send loop')

            await self.disconnect()

    async def _recv_loop(self):
        """
        This loop is constantly putting items on the queue as they're read.
        """
        try:
            while self._connected:
                try:
                    data = await self._recv()
                except asyncio.CancelledError:
                    break
                except (IOError, asyncio.IncompleteReadError) as e:
                    self._log.warning('Server closed the connection: %s', e)
                    await self._recv_queue.put((None, e))
                    await self.disconnect()
                except InvalidChecksumError as e:
                    self._log.warning('Server response had invalid checksum: %s', e)
                    await self._recv_queue.put((None, e))
                except InvalidBufferError as e:
                    self._log.warning('Server response had invalid buffer: %s', e)
                    await self._recv_queue.put((None, e))
                except Exception as e:
                    self._log.exception('Unexpected exception in the receive loop')
                    await self._recv_queue.put((None, e))
                    await self.disconnect()
                else:
                    await self._recv_queue.put((data, None))
        finally:
            await self.disconnect()


    def _init_conn(self):
        """
        This method will be called after `connect` is called.
        After this method finishes, the writer will be drained.

        Subclasses should make use of this if they need to send
        data to Telegram to indicate which connection mode will
        be used.
        """
        if self._codec.tag:
            self._writer.write(self._codec.tag)

    def _send(self, data):
        self._writer.write(self._codec.encode_packet(data))

    async def _recv(self):
        return await self._codec.read_packet(self._reader)

    def __str__(self):
        return '{}:{}/{}'.format(
            self._ip, self._port,
            self.__class__.__name__.replace('Connection', '')
        )


class ObfuscatedConnection(Connection):
    """
    Base class for "obfuscated" connections ("obfuscated2", "mtproto proxy")
    """
    """
    This attribute should be redefined by subclasses
    """
    obfuscated_io = None

    def _init_conn(self):
        self._obfuscation = self.obfuscated_io(self)
        self._writer.write(self._obfuscation.header)

    def _send(self, data):
        self._obfuscation.write(self._codec.encode_packet(data))

    async def _recv(self):
        return await self._codec.read_packet(self._obfuscation)


class PacketCodec(abc.ABC):
    """
    Base class for packet codecs
    """

    """
    This attribute should be re-defined by subclass to define if some
    "magic bytes" should be sent to server right after connection is made to
    signal which protocol will be used
    """
    tag = None

    def __init__(self, connection):
        """
        Codec is created when connection is just made.
        """
        self._conn = connection

    @abc.abstractmethod
    def encode_packet(self, data):
        """
        Encodes single packet and returns encoded bytes.
        """
        raise NotImplementedError

    @abc.abstractmethod
    async def read_packet(self, reader):
        """
        Reads single packet from `reader` object that should have
        `readexactly(n)` method.
        """
        raise NotImplementedError