ss 作用是 f q，因为怕关键词会让文章被干掉，所以用了别名。本文纯粹是从分析技术的角度出发。

因为 ss 现在的版本已经很复杂了，eventloop, 状态机，支持 udp 。为了简单讲原理，我们用 0.9 的版本来说明。

首先上图

这个是 ss 的架构图，用 ss 做代理的时候，一般本地运行个 ss_local.py ，然后服务器再运行 ss_server.py。

client 和 ss_local 主要通过 socks5 协议通信，而 ss_local 和 ss_server 之间就是对称加密的 tcp 数据。

ss_local 主要代码如下

class Socks5Server(SocketServer.StreamRequestHandler): ''' RequesHandlerClass Definition ''' def handle_tcp(self, sock, remote): try: fdset = [sock, remote] while True: r, w, e = select.select(fdset, [], []) # use select I/O multiplexing model if sock in r: # if local socket is ready for reading data = sock.recv(4096) if len(data) <= 0: # received all data break result = send_all(remote, self.encrypt(data)) # send data after encrypting if result < len(data): raise Exception('failed to send all data') if remote in r: # remote socket(proxy) ready for reading data = remote.recv(4096) if len(data) <= 0: break result = send_all(sock, self.decrypt(data)) # send to local socket(application) if result < len(data): raise Exception('failed to send all data') finally: sock.close() remote.close() def encrypt(self, data): return data.translate(encrypt_table) def decrypt(self, data): return data.translate(decrypt_table) def send_encrypt(self, sock, data): sock.send(self.encrypt(data)) def handle(self): try: sock = self.connection # local socket [127.1:port] sock.recv(262) # Sock5 Verification packet sock.send("\\x05\\x00") # Sock5 Response: '0x05' Version 5; '0x00' NO AUTHENTICATION REQUIRED # After Authentication negotiation data = self.rfile.read(4) # Forward request format: VER CMD RSV ATYP (4 bytes) mode = ord(data[1]) # CMD == 0x01 (connect) if mode != 1: logging.warn('mode != 1') return addrtype = ord(data[3]) # indicate destination address type addr_to_send = data[3] if addrtype == 1: # IPv4 addr_ip = self.rfile.read(4) # 4 bytes IPv4 address (big endian) addr = socket.inet_ntoa(addr_ip) addr_to_send += addr_ip elif addrtype == 3: # FQDN (Fully Qualified Domain Name) addr_len = self.rfile.read(1) # Domain name's Length addr = self.rfile.read(ord(addr_len)) # Followed by domain name(e.g. www.google.com) addr_to_send += addr_len + addr else: logging.warn('addr_type not support') # not support return addr_port = self.rfile.read(2) addr_to_send += addr_port # addr_to_send = ATYP + [Length] + dst addr/domain name + port port = struct.unpack('>H', addr_port) # prase the big endian port number. Note: The result is a tuple even if it contains exactly one item. try: reply = "\\x05\\x00\\x00\\x01" # VER REP RSV ATYP reply += socket.inet_aton('0.0.0.0') + struct.pack(">H", 2222) # listening on 2222 on all addresses of the machine, including the loopback(127.0.0.1) self.wfile.write(reply) # response packet # reply immediately if '-6' in sys.argv[1:]: # IPv6 support remote = socket.socket(socket.AF_INET6, socket.SOCK_STREAM) else: remote = socket.socket(socket.AF_INET, socket.SOCK_STREAM) remote.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) # turn off Nagling remote.connect((SERVER, REMOTE_PORT)) self.send_encrypt(remote, addr_to_send) # encrypted logging.info('connecting %s:%d' % (addr, port[0])) except socket.error, e: logging.warn(e) return self.handle_tcp(sock, remote) except socket.error, e: logging.warn(e)

其中和 socks5 交互的部分 handle 在前面 socks5 教程已经讲过，唯一的区别是 send_encrypt函数，每次 send 的时候都会 encrypt 一次，拿到数据后再 decrypt 一次。

ss_server 的代码也和 local 差不多，少了和 socks5 打交道

def handle(self): try: sock = self.connection addrtype = ord(self.decrypt(sock.recv(1))) # receive addr type if addrtype == 1: addr = socket.inet_ntoa(self.decrypt(self.rfile.read(4))) # get dst addr elif addrtype == 3: addr = self.decrypt( self.rfile.read(ord(self.decrypt(sock.recv(1))))) # read 1 byte of len, then get 'len' bytes name else: # not support logging.warn('addr_type not support') return port = struct.unpack('>H', self.decrypt(self.rfile.read(2))) # get dst port into small endian try: logging.info('connecting %s:%d' % (addr, port[0])) remote = socket.socket(socket.AF_INET, socket.SOCK_STREAM) remote.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) remote.connect((addr, port[0])) # connect to dst except socket.error, e: # Connection refused logging.warn(e) return self.handle_tcp(sock, remote) except socket.error, e: logging.warn(e)

本质上 ss 很简单，就是做了流量的转发，只不过为了避免流量被检测，加密并且还有混淆的功能。