From 1bc0e60cfd3091a184a7f102893f0ccc43083b33 Mon Sep 17 00:00:00 2001 From: breakwa11 Date: Tue, 20 Oct 2015 18:41:35 +0800 Subject: [PATCH] merge manyuser branch --- shadowsocks/common.py | 2 +- shadowsocks/obfs.py | 21 +++- shadowsocks/obfsplugin/http_simple.py | 4 +- shadowsocks/obfsplugin/plain.py | 9 +- shadowsocks/obfsplugin/verify_simple.py | 160 +++++++++++++++++++++--- shadowsocks/shell.py | 9 +- shadowsocks/tcprelay.py | 41 +++++- shadowsocks/udprelay.py | 23 +++- 8 files changed, 232 insertions(+), 37 deletions(-) diff --git a/shadowsocks/common.py b/shadowsocks/common.py index cc88d5d..2b11b9f 100644 --- a/shadowsocks/common.py +++ b/shadowsocks/common.py @@ -160,7 +160,7 @@ def pre_parse_header(data): 'encryption method') return None data = data[rand_data_size + 3:] - elif datatype == 0x88: + elif datatype == 0x88 or (~datatype & 0xff) == 0x88: if len(data) <= 7 + 7: return None data_size = struct.unpack('>H', data[1:3])[0] diff --git a/shadowsocks/obfs.py b/shadowsocks/obfs.py index 69eed0b..92de548 100644 --- a/shadowsocks/obfs.py +++ b/shadowsocks/obfs.py @@ -27,19 +27,28 @@ from shadowsocks.obfsplugin import plain, http_simple, verify_simple method_supported = {} -method_supported.update(plain.obfs) -method_supported.update(http_simple.obfs) -method_supported.update(verify_simple.obfs) +method_supported.update(plain.obfs_map) +method_supported.update(http_simple.obfs_map) +method_supported.update(verify_simple.obfs_map) -class Obfs(object): +class server_info(object): + def __init__(self, data): + self.data = data + +class obfs(object): def __init__(self, method): self.method = method self._method_info = self.get_method_info(method) if self._method_info: self.obfs = self.get_obfs(method) else: - logging.error('method %s not supported' % method) - sys.exit(1) + raise Exception('method %s not supported' % method) + + def init_data(self): + return self.obfs.init_data() + + def set_server_info(self, server_info): + return self.obfs.set_server_info(server_info) def get_method_info(self, method): method = method.lower() diff --git a/shadowsocks/obfsplugin/http_simple.py b/shadowsocks/obfsplugin/http_simple.py index f9d0476..75e9294 100644 --- a/shadowsocks/obfsplugin/http_simple.py +++ b/shadowsocks/obfsplugin/http_simple.py @@ -25,8 +25,8 @@ import binascii import base64 import datetime -from shadowsocks.obfsplugin import plain from shadowsocks import common +from shadowsocks.obfsplugin import plain from shadowsocks.common import to_bytes, to_str, ord def create_http_obfs(method): @@ -41,7 +41,7 @@ def create_tls_obfs(method): def create_random_head_obfs(method): return random_head(method) -obfs = { +obfs_map = { 'http_simple': (create_http_obfs,), 'http_simple_compatible': (create_http_obfs,), 'http2_simple': (create_http2_obfs,), diff --git a/shadowsocks/obfsplugin/plain.py b/shadowsocks/obfsplugin/plain.py index a40b993..cc602fa 100644 --- a/shadowsocks/obfsplugin/plain.py +++ b/shadowsocks/obfsplugin/plain.py @@ -25,13 +25,20 @@ import logging def create_obfs(method): return plain(method) -obfs = { +obfs_map = { 'plain': (create_obfs,), } class plain(object): def __init__(self, method): self.method = method + self.server_info = None + + def init_data(self): + return b'' + + def set_server_info(self, server_info): + self.server_info = server_info def client_pre_encrypt(self, buf): return buf diff --git a/shadowsocks/obfsplugin/verify_simple.py b/shadowsocks/obfsplugin/verify_simple.py index f89e743..10ba8f7 100644 --- a/shadowsocks/obfsplugin/verify_simple.py +++ b/shadowsocks/obfsplugin/verify_simple.py @@ -25,16 +25,22 @@ import binascii import base64 import datetime import struct +import zlib -from shadowsocks.obfsplugin import plain +import shadowsocks from shadowsocks import common +from shadowsocks.obfsplugin import plain from shadowsocks.common import to_bytes, to_str, ord def create_verify_obfs(method): return verify_simple(method) -obfs = { +def create_verify_deflate(method): + return verify_deflate(method) + +obfs_map = { 'verify_simple': (create_verify_obfs,), + 'verify_deflate': (create_verify_deflate,), } def match_begin(str1, str2): @@ -43,9 +49,64 @@ def match_begin(str1, str2): return True return False -class verify_simple(plain.plain): +class sub_encode_obfs(object): + def __init__(self): + self.sub_obfs = None + +class verify_base(plain.plain): def __init__(self, method): + super(verify_base, self).__init__(method) self.method = method + self.sub_obfs = None + + def init_data(self): + return sub_encode_obfs() + + def set_server_info(self, server_info): + try: + if server_info.param: + sub_param = '' + param_list = server_info.param.split(',', 1) + if len(param_list) > 1: + self.sub_obfs = shadowsocks.obfs.obfs(param_list[0]) + sub_param = param_list[1] + else: + self.sub_obfs = shadowsocks.obfs.obfs(server_info.param) + if server_info.data.sub_obfs is None: + server_info.data.sub_obfs = self.sub_obfs.init_data() + _server_info = shadowsocks.obfs.server_info(server_info.data.sub_obfs) + _server_info.host = server_info.host + _server_info.port = server_info.port + _server_info.tcp_mss = server_info.tcp_mss + _server_info.param = sub_param + self.sub_obfs.set_server_info(_server_info) + except Exception as e: + shadowsocks.shell.print_exception(e) + self.server_info = server_info + + def client_encode(self, buf): + if self.sub_obfs is not None: + return self.sub_obfs.client_encode(buf) + return buf + + def client_decode(self, buf): + if self.sub_obfs is not None: + return self.sub_obfs.client_decode(buf) + return (buf, False) + + def server_encode(self, buf): + if self.sub_obfs is not None: + return self.sub_obfs.server_encode(buf) + return buf + + def server_decode(self, buf): + if self.sub_obfs is not None: + return self.sub_obfs.server_decode(buf) + return (buf, True, False) + +class verify_simple(verify_base): + def __init__(self, method): + super(verify_simple, self).__init__(method) self.recv_buf = b'' self.unit_len = 8100 self.decrypt_packet_num = 0 @@ -69,13 +130,6 @@ class verify_simple(plain.plain): ret += self.pack_data(buf) return ret - def client_encode(self, buf): - return buf - - def client_decode(self, buf): - # (buffer_to_recv, is_need_to_encode_and_send_back) - return (buf, False) - def client_post_decrypt(self, buf): if self.raw_trans: return buf @@ -117,13 +171,6 @@ class verify_simple(plain.plain): ret += self.pack_data(buf) return ret - def server_encode(self, buf): - return buf - - def server_decode(self, buf): - # (buffer_to_recv, is_need_decrypt, is_need_to_encode_and_send_back) - return (buf, True, False) - def server_post_decrypt(self, buf): if self.raw_trans: return buf @@ -157,3 +204,82 @@ class verify_simple(plain.plain): self.decrypt_packet_num += 1 return out_buf +class verify_deflate(verify_base): + def __init__(self, method): + super(verify_deflate, self).__init__(method) + self.recv_buf = b'' + self.unit_len = 32700 + self.decrypt_packet_num = 0 + self.raw_trans = False + + def pack_data(self, buf): + if len(buf) == 0: + return b'' + data = zlib.compress(buf) + data = struct.pack('>H', len(data)) + data[2:] + return data + + def client_pre_encrypt(self, buf): + ret = b'' + while len(buf) > self.unit_len: + ret += self.pack_data(buf[:self.unit_len]) + buf = buf[self.unit_len:] + ret += self.pack_data(buf) + return ret + + def client_post_decrypt(self, buf): + if self.raw_trans: + return buf + self.recv_buf += buf + out_buf = b'' + while len(self.recv_buf) > 2: + length = struct.unpack('>H', self.recv_buf[:2])[0] + if length >= 32768: + self.raw_trans = True + self.recv_buf = b'' + if self.decrypt_packet_num == 0: + return None + else: + raise Exception('server_post_decrype data error') + if length > len(self.recv_buf): + break + + out_buf += zlib.decompress(b'x\x9c' + self.recv_buf[2:length]) + self.recv_buf = self.recv_buf[length:] + + if out_buf: + self.decrypt_packet_num += 1 + return out_buf + + def server_pre_encrypt(self, buf): + ret = b'' + while len(buf) > self.unit_len: + ret += self.pack_data(buf[:self.unit_len]) + buf = buf[self.unit_len:] + ret += self.pack_data(buf) + return ret + + def server_post_decrypt(self, buf): + if self.raw_trans: + return buf + self.recv_buf += buf + out_buf = b'' + while len(self.recv_buf) > 2: + length = struct.unpack('>H', self.recv_buf[:2])[0] + if length >= 32768: + self.raw_trans = True + self.recv_buf = b'' + if self.decrypt_packet_num == 0: + return None + else: + raise Exception('server_post_decrype data error') + if length > len(self.recv_buf): + break + + out_buf += zlib.decompress(b'\x78\x9c' + self.recv_buf[2:length]) + self.recv_buf = self.recv_buf[length:] + + if out_buf: + self.decrypt_packet_num += 1 + return out_buf + diff --git a/shadowsocks/shell.py b/shadowsocks/shell.py index e5f5bb5..66e38bb 100644 --- a/shadowsocks/shell.py +++ b/shadowsocks/shell.py @@ -130,11 +130,11 @@ def get_config(is_local): logging.basicConfig(level=logging.INFO, format='%(levelname)-s: %(message)s') if is_local: - shortopts = 'hd:s:b:p:k:l:m:c:t:vq' + shortopts = 'hd:s:b:p:k:l:m:o:c:t:vq' longopts = ['help', 'fast-open', 'pid-file=', 'log-file=', 'user=', 'version'] else: - shortopts = 'hd:s:p:k:m:c:t:vq' + shortopts = 'hd:s:p:k:m:o:c:t:vq' longopts = ['help', 'fast-open', 'pid-file=', 'log-file=', 'workers=', 'forbidden-ip=', 'user=', 'manager-address=', 'version'] try: @@ -168,6 +168,8 @@ def get_config(is_local): config['server'] = to_str(value) elif key == '-m': config['method'] = to_str(value) + elif key == '-o': + config['obfs'] = to_str(value) elif key == '-b': config['local_address'] = to_str(value) elif key == '-v': @@ -217,6 +219,7 @@ def get_config(is_local): config['password'] = to_bytes(config.get('password', b'')) config['method'] = to_str(config.get('method', 'aes-256-cfb')) config['obfs'] = to_str(config.get('obfs', 'plain')) + config['obfs_param'] = to_str(config.get('obfs_param', '')) config['port_password'] = config.get('port_password', None) config['timeout'] = int(config.get('timeout', 300)) config['fast_open'] = config.get('fast_open', False) @@ -286,6 +289,7 @@ Proxy options: -l LOCAL_PORT local port, default: 1080 -k PASSWORD password -m METHOD encryption method, default: aes-256-cfb + -o OBFS obfsplugin, default: http_simple -t TIMEOUT timeout in seconds, default: 300 --fast-open use TCP_FASTOPEN, requires Linux 3.7+ @@ -315,6 +319,7 @@ Proxy options: -p SERVER_PORT server port, default: 8388 -k PASSWORD password -m METHOD encryption method, default: aes-256-cfb + -o OBFS obfsplugin, default: http_simple -t TIMEOUT timeout in seconds, default: 300 --fast-open use TCP_FASTOPEN, requires Linux 3.7+ --workers WORKERS number of workers, available on Unix/Linux diff --git a/shadowsocks/tcprelay.py b/shadowsocks/tcprelay.py index 99ac9b3..2bb973a 100644 --- a/shadowsocks/tcprelay.py +++ b/shadowsocks/tcprelay.py @@ -117,7 +117,16 @@ class TCPRelayHandler(object): self._encryptor = encrypt.Encryptor(config['password'], config['method']) self._encrypt_correct = True - self._obfs = obfs.Obfs(config['obfs']) + self._obfs = obfs.obfs(config['obfs']) + if server.obfs_data is None: + server.obfs_data = self._obfs.init_data() + server_info = obfs.server_info(server.obfs_data) + server_info.host = config['server'] + server_info.port = server._listen_port + server_info.tcp_mss = 1440 + server_info.param = config['obfs_param'] + self._obfs.set_server_info(server_info) + self._fastopen_connected = False self._data_to_write_to_local = [] self._data_to_write_to_remote = [] @@ -139,6 +148,7 @@ class TCPRelayHandler(object): self._server) self.last_activity = 0 self._update_activity() + self._server.add_connection(1) def __hash__(self): # default __hash__ is id / 16 @@ -276,6 +286,10 @@ class TCPRelayHandler(object): shell.print_exception(e) self.destroy() return False + except Exception as e: + shell.print_exception(e) + self.destroy() + return False if uncomplete: if sock == self._local_sock: self._data_to_write_to_local.append(data) @@ -390,7 +404,7 @@ class TCPRelayHandler(object): if self._is_local: header_result = parse_header(data) else: - if data is None or FORCE_NEW_PROTOCOL and common.ord(data[0]) != 0x88: + if data is None or FORCE_NEW_PROTOCOL and common.ord(data[0]) != 0x88 and (~common.ord(data[0]) & 0xff) != 0x88: data = self._handel_protocol_error(self._client_address, ogn_data) data = pre_parse_header(data) if data is None: @@ -467,7 +481,7 @@ class TCPRelayHandler(object): remote_sock.setblocking(False) if self._remote_udp: - pass + remote_sock_v6.setblocking(False) else: remote_sock.setsockopt(socket.SOL_TCP, socket.TCP_NODELAY, 1) return remote_sock @@ -515,7 +529,9 @@ class TCPRelayHandler(object): except (OSError, IOError) as e: if eventloop.errno_from_exception(e) == \ errno.EINPROGRESS: - pass + pass # always goto here + else: + raise e self._loop.add(remote_sock, eventloop.POLL_ERR | eventloop.POLL_OUT, self._server) @@ -721,13 +737,19 @@ class TCPRelayHandler(object): logging.debug('destroy') if self._remote_sock: logging.debug('destroying remote') - self._loop.remove(self._remote_sock) + try: + self._loop.remove(self._remote_sock) + except Exception as e: + pass del self._fd_to_handlers[self._remote_sock.fileno()] self._remote_sock.close() self._remote_sock = None if self._remote_sock_v6: logging.debug('destroying remote') - self._loop.remove(self._remote_sock_v6) + try: + self._loop.remove(self._remote_sock_v6) + except Exception as e: + pass del self._fd_to_handlers[self._remote_sock_v6.fileno()] self._remote_sock_v6.close() self._remote_sock_v6 = None @@ -739,6 +761,7 @@ class TCPRelayHandler(object): self._local_sock = None self._dns_resolver.remove_callback(self._handle_dns_resolved) self._server.remove_handler(self) + self._server.add_connection(-1) class TCPRelay(object): def __init__(self, config, dns_resolver, is_local, stat_callback=None): @@ -750,6 +773,8 @@ class TCPRelay(object): self._fd_to_handlers = {} self.server_transfer_ul = 0 self.server_transfer_dl = 0 + self.server_connections = 0 + self.obfs_data = None self._timeout = config['timeout'] self._timeouts = [] # a list for all the handlers @@ -802,6 +827,10 @@ class TCPRelay(object): self._timeouts[index] = None del self._handler_to_timeouts[hash(handler)] + def add_connection(self, val): + self.server_connections += val + logging.debug('server port %5d connections = %d' % (self._listen_port, self.server_connections,)) + def update_activity(self, handler, data_len): if data_len and self._stat_callback: self._stat_callback(self._listen_port, data_len) diff --git a/shadowsocks/udprelay.py b/shadowsocks/udprelay.py index d7626c9..d3696d9 100644 --- a/shadowsocks/udprelay.py +++ b/shadowsocks/udprelay.py @@ -1095,6 +1095,12 @@ class UDPRelay(object): self._sockets.add(client.fileno()) self._eventloop.add(client, eventloop.POLL_IN, self) + logging.debug('UDP port %5d sockets %d' % (self._listen_port, len(self._sockets))) + + logging.info('UDP data to %s:%d from %s:%d' % + (common.to_str(server_addr), server_port, + r_addr[0], r_addr[1])) + if self._is_local: data = encrypt.encrypt_all(self._password, self._method, 1, data) if not data: @@ -1254,11 +1260,21 @@ class UDPRelay(object): if sock == self._server_socket: if event & eventloop.POLL_ERR: logging.error('UDP server_socket err') - self._handle_server() + try: + self._handle_server() + except Exception as e: + shell.print_exception(e) + if self._config['verbose']: + traceback.print_exc() elif sock and (fd in self._sockets): if event & eventloop.POLL_ERR: logging.error('UDP client_socket err') - self._handle_client(sock) + try: + self._handle_client(sock) + except Exception as e: + shell.print_exception(e) + if self._config['verbose']: + traceback.print_exc() else: if sock: handler = self._fd_to_handlers.get(fd, None) @@ -1275,7 +1291,10 @@ class UDPRelay(object): for sock in self._sockets: sock.close() logging.info('closed UDP port %d', self._listen_port) + before_sweep_size = len(self._sockets) self._cache.sweep() + if before_sweep_size != len(self._sockets): + logging.debug('UDP port %5d sockets %d' % (self._listen_port, len(self._sockets))) self._client_fd_to_server_addr.sweep() self._sweep_timeout()