shadowsocksr/shadowsocks/tcprelay.py

#!/usr/bin/python
# -*- coding: utf-8 -*-

# Copyright (c) 2014 clowwindy
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.

import time
import socket
import logging
import encrypt
import errno
import eventloop
from common import parse_header

CMD_CONNECT = 1

# local:
# stage 0 init
# stage 1 hello received, hello sent
# stage 4 addr received, reply sent
# stage 5 remote connected

# remote:
# stage 0 init
# stage 4 addr received, reply sent
# stage 5 remote connected

STAGE_INIT = 0
STAGE_HELLO = 1
STAGE_REPLY = 4
STAGE_STREAM = 5

# stream direction
STREAM_UP = 0
STREAM_DOWN = 1

# stream status
STATUS_WAIT_INIT = 0
STATUS_WAIT_READING = 1
STATUS_WAIT_WRITING = 2
STATUS_WAIT_READWRITING = STATUS_WAIT_READING | STATUS_WAIT_WRITING

BUF_SIZE = 8 * 1024


class TCPRelayHandler(object):
    def __init__(self, fd_to_handlers, loop, local_sock, config, is_local):
        self._fd_to_handlers = fd_to_handlers
        self._loop = loop
        self._local_sock = local_sock
        self._remote_sock = None
        self._config = config
        self._is_local = is_local
        self._stage = STAGE_INIT
        self._encryptor = encrypt.Encryptor(config['password'],
                                            config['method'])
        self._data_to_write_to_local = []
        self._data_to_write_to_remote = []
        self._upstream_status = STATUS_WAIT_READING
        self._downstream_status = STATUS_WAIT_INIT
        fd_to_handlers[local_sock.fileno()] = self
        local_sock.setblocking(False)
        local_sock.setsockopt(socket.SOL_TCP, socket.TCP_NODELAY, 1)
        loop.add(local_sock, eventloop.POLL_IN | eventloop.POLL_ERR)

    def update_stream(self, stream, status):
        dirty = False
        if stream == STREAM_DOWN:
            if self._downstream_status != status:
                self._downstream_status = status
                dirty = True
        elif stream == STREAM_UP:
            if self._upstream_status != status:
                self._upstream_status = status
                dirty = True
        if dirty:
            if self._local_sock:
                event = eventloop.POLL_ERR
                if self._downstream_status & STATUS_WAIT_WRITING:
                    event |= eventloop.POLL_OUT
                if self._upstream_status & STATUS_WAIT_READING:
                    event |= eventloop.POLL_IN
                self._loop.modify(self._local_sock, event)
            if self._remote_sock:
                event = eventloop.POLL_ERR
                if self._downstream_status & STATUS_WAIT_READING:
                    event |= eventloop.POLL_IN
                if self._upstream_status & STATUS_WAIT_WRITING:
                    event |= eventloop.POLL_OUT
                self._loop.modify(self._remote_sock, event)

    def write_all_to_sock(self, data, sock):
        if not data or not sock:
            return
        uncomplete = False
        try:
            l = len(data)
            s = sock.send(data)
            if s < l:
                data = data[s:]
                uncomplete = True
        except (OSError, IOError) as e:
            error_no = eventloop.errno_from_exception(e)
            if error_no in (errno.EAGAIN, errno.EINPROGRESS):
                uncomplete = True
            else:
                logging.error(e)
                self.destroy()
        if uncomplete:
            if sock == self._local_sock:
                self._data_to_write_to_local.append(data)
                self.update_stream(STREAM_DOWN, STATUS_WAIT_WRITING)
            elif sock == self._remote_sock:
                self._data_to_write_to_remote.append(data)
                self.update_stream(STREAM_UP, STATUS_WAIT_WRITING)
            else:
                logging.error('write_all_to_sock:unknown socket')
        else:
            if sock == self._local_sock:
                self.update_stream(STREAM_DOWN, STATUS_WAIT_READING)
            elif sock == self._remote_sock:
                self.update_stream(STREAM_UP, STATUS_WAIT_READING)
            else:
                logging.error('write_all_to_sock:unknown socket')

    def on_local_read(self):
        # TODO update timeout
        if not self._local_sock:
            return
        is_local = self._is_local
        data = None
        try:
            data = self._local_sock.recv(BUF_SIZE)
        except (OSError, IOError) as e:
            if eventloop.errno_from_exception(e) in \
                    (errno.ETIMEDOUT, errno.EAGAIN):
                return
        if not data:
            self.destroy()
            return
        if not is_local:
            data = self._encryptor.decrypt(data)
            if not data:
                return
        if self._stage == STAGE_STREAM:
            if self._is_local:
                data = self._encryptor.encrypt(data)
            self.write_all_to_sock(data, self._remote_sock)
            return
        elif is_local and self._stage == STAGE_INIT:
            # TODO check auth method
            self.write_all_to_sock('\x05\00', self._local_sock)
            self._stage = STAGE_HELLO
            return
        elif self._stage == STAGE_REPLY:
            if is_local:
                data = self._encryptor.encrypt(data)
            self._data_to_write_to_remote.append(data)
        elif (is_local and self._stage == STAGE_HELLO) or \
                (not is_local and self._stage == STAGE_INIT):
            try:
                if is_local:
                    cmd = ord(data[1])
                    # TODO check cmd == 1
                    assert cmd == CMD_CONNECT
                    # just trim VER CMD RSV
                    data = data[3:]
                header_result = parse_header(data)
                if header_result is None:
                    raise Exception('can not parse header')
                addrtype, remote_addr, remote_port, header_length =\
                    header_result
                logging.debug('connecting %s:%d' % (remote_addr, remote_port))
                if is_local:
                    # forward address to remote
                    self.write_all_to_sock('\x05\x00\x00\x01' +
                                           '\x00\x00\x00\x00\x10\x10',
                                           self._local_sock)
                    data_to_send = self._encryptor.encrypt(data)
                    self._data_to_write_to_remote.append(data_to_send)
                    remote_addr = self._config['server']
                    remote_port = self._config['server_port']
                else:
                    if len(data) > header_length:
                        self._data_to_write_to_remote.append(
                            data[header_length:])

                # TODO async DNS
                addrs = socket.getaddrinfo(remote_addr, remote_port, 0,
                                           socket.SOCK_STREAM, socket.SOL_TCP)
                if len(addrs) == 0:
                    raise Exception("can't get addrinfo for %s:%d" %
                                    (remote_addr, remote_port))
                af, socktype, proto, canonname, sa = addrs[0]
                remote_sock = socket.socket(af, socktype, proto)
                self._remote_sock = remote_sock
                self._fd_to_handlers[remote_sock.fileno()] = self
                remote_sock.setblocking(False)
                remote_sock.setsockopt(socket.SOL_TCP, socket.TCP_NODELAY, 1)
                # TODO support TCP fast open
                try:
                    remote_sock.connect(sa)
                except (OSError, IOError) as e:
                    if eventloop.errno_from_exception(e) == errno.EINPROGRESS:
                        pass
                self._loop.add(remote_sock,
                               eventloop.POLL_ERR | eventloop.POLL_OUT)

                self._stage = STAGE_REPLY
                self.update_stream(STREAM_UP, STATUS_WAIT_READWRITING)
                self.update_stream(STREAM_DOWN, STATUS_WAIT_READING)
                return
            except Exception:
                import traceback
                traceback.print_exc()
                # TODO use logging when debug completed
                self.destroy()

    def on_remote_read(self):
        # TODO update timeout
        data = None
        try:
            data = self._remote_sock.recv(BUF_SIZE)
        except (OSError, IOError) as e:
            if eventloop.errno_from_exception(e) in \
                    (errno.ETIMEDOUT, errno.EAGAIN):
                return
        if not data:
            self.destroy()
            return
        if self._is_local:
            data = self._encryptor.decrypt(data)
        else:
            data = self._encryptor.encrypt(data)
        try:
            self.write_all_to_sock(data, self._local_sock)
        except Exception:
            import traceback
            traceback.print_exc()
            # TODO use logging when debug completed
            self.destroy()

    def on_local_write(self):
        if self._data_to_write_to_local:
            data = ''.join(self._data_to_write_to_local)
            self._data_to_write_to_local = []
            self.write_all_to_sock(data, self._local_sock)
        else:
            self.update_stream(STREAM_DOWN, STATUS_WAIT_READING)

    def on_remote_write(self):
        self._stage = STAGE_STREAM
        if self._data_to_write_to_remote:
            data = ''.join(self._data_to_write_to_remote)
            self._data_to_write_to_remote = []
            self.write_all_to_sock(data, self._remote_sock)
        else:
            self.update_stream(STREAM_UP, STATUS_WAIT_READING)

    def on_local_error(self):
        logging.error(eventloop.get_sock_error(self._local_sock))
        self.destroy()

    def on_remote_error(self):
        logging.error(eventloop.get_sock_error(self._remote_sock))
        self.destroy()

    def handle_event(self, sock, event):
        # order is important
        if sock == self._remote_sock:
            if event & eventloop.POLL_IN:
                self.on_remote_read()
            if event & eventloop.POLL_OUT:
                self.on_remote_write()
            if event & eventloop.POLL_ERR:
                self.on_remote_error()
        elif sock == self._local_sock:
            if event & eventloop.POLL_IN:
                self.on_local_read()
            if event & eventloop.POLL_OUT:
                self.on_local_write()
            if event & eventloop.POLL_ERR:
                self.on_local_error()
        else:
            logging.warn('unknown socket')

    def destroy(self):
        logging.debug('destroy')
        if self._remote_sock:
            self._loop.remove(self._remote_sock)
            del self._fd_to_handlers[self._remote_sock.fileno()]
            self._remote_sock.close()
            self._remote_sock = None
        if self._local_sock:
            self._loop.remove(self._local_sock)
            del self._fd_to_handlers[self._local_sock.fileno()]
            self._local_sock.close()
            self._local_sock = None


class TCPRelay(object):
    def __init__(self, config, is_local):
        self._config = config
        self._is_local = is_local
        self._closed = False
        self._eventloop = None
        self._fd_to_handlers = {}
        self._last_time = time.time()

        if is_local:
            listen_addr = config['local_address']
            listen_port = config['local_port']
        else:
            listen_addr = config['server']
            listen_port = config['server_port']

        addrs = socket.getaddrinfo(listen_addr, listen_port, 0,
                                   socket.SOCK_STREAM, socket.SOL_TCP)
        if len(addrs) == 0:
            raise Exception("can't get addrinfo for %s:%d" %
                            (listen_addr, listen_port))
        af, socktype, proto, canonname, sa = addrs[0]
        server_socket = socket.socket(af, socktype, proto)
        server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        server_socket.bind(sa)
        server_socket.setblocking(False)
        server_socket.listen(1024)
        self._server_socket = server_socket

    def add_to_loop(self, loop):
        if self._closed:
            raise Exception('already closed')
        self._eventloop = loop
        loop.add_handler(self._handle_events)

        self._eventloop.add(self._server_socket,
                            eventloop.POLL_IN | eventloop.POLL_ERR)

    def _handle_events(self, events):
        for sock, fd, event in events:
            if sock:
                logging.debug('fd %d %s', fd,
                              eventloop.EVENT_NAMES.get(event, event))
            if sock == self._server_socket:
                if event & eventloop.POLL_ERR:
                    # TODO
                    raise Exception('server_socket error')
                try:
                    logging.debug('accept')
                    conn = self._server_socket.accept()
                    TCPRelayHandler(self._fd_to_handlers, self._eventloop,
                                    conn[0], self._config, self._is_local)
                except (OSError, IOError) as e:
                    error_no = eventloop.errno_from_exception(e)
                    if error_no in (errno.EAGAIN, errno.EINPROGRESS):
                        continue
                    else:
                        logging.error(e)
            else:
                if sock:
                    handler = self._fd_to_handlers.get(fd, None)
                    if handler:
                        handler.handle_event(sock, event)
                else:
                    logging.warn('poll removed fd')

            now = time.time()
            if now - self._last_time > 5:
                # TODO sweep timeouts
                self._last_time = now

    def close(self):
        self._closed = True
        self._server_socket.close()