SOCKS5协议详解及Python简单实现

SOCKS协议

SOCKS是一种网络传输协议，主要用于客户端与外网服务器之间通讯的中间传递。SOCKS是”SOCKet Secure”的缩写。

当防火墙后的客户端要访问外部的服务器时，就跟SOCKS代理服务器连接。这个代理服务器控制客户端访问外网的资格，允许的话，就将客户端的请求发往外部的服务器。

这个协议最初由David Koblas开发，而后由NEC的Ying-Da Lee将其扩展到SOCKS4。最新协议是SOCKS5，与前一版本相比，增加支持UDP、验证，以及IPv6。

根据OSI模型，SOCKS是会话层的协议，位于表示层与传输层之间。

SOCKS协议不提供加密。

SOCKS5协议

协商

1. 客户端连接到 SOCKS 服务端，发送的协议版本与认证方法数据包格式：

+----+----------+----------+
|VER | NMETHODS | METHODS  |
+----+----------+----------+
| 1  |    1     | 1 to 255 |
+----+----------+----------+

• VER 是指协议版本，因为是 socks5，所以值是 0x05,一个字节
• NMETHODS 是指有多少个可以使用的方法，也就是客户端支持的认证方法，一个字节
• METHODS 就是方法值，1-255个字节，有多少个方法就有多少个byte

1. 服务端要从给定的方法列表中选择一个方法并返回选择报文：

+----+--------+
|VER | METHOD |
+----+--------+
| 1  |   1    |
+----+--------+

VER与METHOD的取值同上

目前已定义方法如下：

• 0x00 NO AUTHENTICATION REQUIRED 不需要认证
• 0x01 GSSAPI 参考：GSSAPI
• 0x02 USERNAME/PASSWORD 用户名密码认证
• 0x03 to 0x7F IANA ASSIGNED 一般不用。INNA保留。
• 0x80 to 0xFE RESERVED FOR PRIVATE METHODS 保留作私有用处。
• 0xFF NO ACCEPTABLE METHODS 不接受任何方法/没有合适的方法

请求

一旦认证方法对应的协商完成，客户端就可以发送请求细节了。如果认证方法为了完整性或者可信性的校验，需要对后续请求报文进行封装，则后续请求报文都要按照对应规定进行封装。

SOCKS 请求为如下格式：

+----+-----+-------+------+----------+----------+
|VER | CMD |  RSV  | ATYP | DST.ADDR | DST.PORT |
+----+-----+-------+------+----------+----------+
| 1  |  1  | X'00' |  1   | Variable |    2     |
+----+-----+-------+------+----------+----------+

• VER 还是版本，取值是 0x05
• CMD 是指要做啥，取值如下：
  • CONNECT 0x01 连接
  • BIND 0x02 端口监听(也就是在Server上监听一个端口)
  • UDP ASSOCIATE 0x03 使用UDP
• RSV 是保留位，值是 0x00
• ATYP 是目标地址类型，有如下取值：
  • 0x01 IPv4
  • 0x03 域名
  • 0x04 IPv6
• DST.ADDR 就是目标地址的值了，如果是IPv4，那么就是4 bytes，如果是IPv6那么就是16 bytes，如果是域名，那么第一个字节代表接下来有多少个字节是表示目标地址
• DST.PORT 两个字节代表端口号

SOCKS 服务端会根据请求类型和源、目标地址，执行对应操作，并且返回对应的一个或多个报文信息。

回复

客户端与服务端建立连接并完成认证之后就会发送请求信息，服务端执行对应请求并返回如下格式的报文：

+----+-----+-------+------+----------+----------+
|VER | REP |  RSV  | ATYP | BND.ADDR | BND.PORT |
+----+-----+-------+------+----------+----------+
| 1  |  1  | X'00' |  1   | Variable |    2     |
+----+-----+-------+------+----------+----------+

• VER 还是版本，值是 0x05
• REP 是状态码，取值如下：
  • 0x00 succeeded
  • 0x01 general SOCKS server failure
  • 0x02 connection not allowed by ruleset
  • 0x03 Network unreachable
  • 0x04 Host unreachable
  • 0x05 Connection refused
  • 0x06 TTL expired
  • 0x07 Command not supported
  • 0x08 Address type not supported
  • 0x09 to 0xff unassigned
• RSV 保留位，取值为 0x00
• ATYP 是目标地址类型，有如下取值：
  • 0x01 IPv4
  • 0x03 域名
  • 0x04 IPv6
• DST.ADDR 就是目标地址的值了，如果是IPv4，那么就是4 bytes，如果是IPv6那么就是16 bytes，如果是域名，那么第一个字节代表接下来有多少个字节是表示目标地址
• DST.PORT 两个字节代表端口号

数据转发

• 到了这个阶段基本就是数据转发了，tcp就直接转发，udp还须要做点工作。
  • 当一个响应（REP值不为X00）指示失败的时候，SOCKS服务器必须在发送这个响应后立刻断开这条TCP连接。这必须发生在检测到引起失败的原因之后的10秒内。
  • 如果响应码（REP值为X00）指示成功，而且这次请求是BIND或者CONNECT，那么客户端可以立刻开始数据传输。如果选择的认证方法支持针对完整性、认证或私密性目的封装，要传输的数据应该包装在所依赖的认证方法的封装中。同样地，当针对客户端的响应数据到达SOCKS服务器的时候，服务器也必须根据使用的认证方法来封装数据。
• 一个基于UDP的客户端必须将它的数据报发送到UDP中继服务器的指定端口——该端口在针对UDP ASSOCIATE的响应中的BND.PORT域指明。如果选择的认证方法提供了针对认证、完整性或私密性目的的封装，数据报必须使用恰当的封装方式进行包装。每一个UDP数据报都随身携带了一个UDP请求头：

+----+------+------+----------+----------+----------+
|RSV | FRAG | ATYP | DST.ADDR | DST.PORT |   DATA   |
+----+------+------+----------+----------+----------+
| 2  |  1   |  1   | Variable |    2     | Variable |
+----+------+------+----------+----------+----------+

• RSV　保留字段，应当置为 X‘0000’
• FRAG　当前帧序号
• ATYP　地址类型
  • IPV4　X‘01’
  • 域名　X‘03’
  • IPV6　X‘04’
• DST.ADDR　目的地址
• DST.PORT　目的端口
• DATA　用户数据

当一个UDP中继服务器决定中继一个UDP数据报的时候，它会默默的做——不会给客户端返回任何通知。同样的，如果它不能中继数据报那它就会丢弃数据报。当一个UDP中继服务器从远端主机收到一个响应数据报文的时候，它必须根据使用上述的UDP请求头对该响应报文进行封装，然后再进行所依赖的认证方法的封装处理。

SOCKS5 服务器（Python）

不需要认证：

import select
import socket
import struct
from socketserver import StreamRequestHandler, ThreadingTCPServer

SOCKS_VERSION = 5


def generate_failed_reply(address_type, error_number):
    return struct.pack("!BBBBIH", SOCKS_VERSION, error_number, 0, address_type, 0, 0)


def exchange_loop(client, remote):
    while True:
        # 等待数据
        r, w, e = select.select([client, remote], [], [])
        if client in r:
            data = client.recv(4096)
            if remote.send(data) <= 0:
                break
        if remote in r:
            data = remote.recv(4096)
            if client.send(data) <= 0:
                break


class SocksProxy(StreamRequestHandler):
    def handle(self):
        print('Accepting connection from {}'.format(self.client_address))

        # 协商
        # 从客户端读取并解包两个字节的数据
        header = self.connection.recv(2)
        version, nmethods = struct.unpack("!BB", header)

        # 设置socks5协议，METHODS字段的数目大于0
        assert version == SOCKS_VERSION
        assert nmethods > 0

        # 接受支持的方法
        methods = self.get_available_methods(nmethods)

        # 无需认证
        if 0 not in set(methods):
            self.server.close_request(self.request)
            return

        # 发送协商响应数据包
        self.connection.sendall(struct.pack("!BB", SOCKS_VERSION, 0))

        # 请求
        version, cmd, _, address_type = struct.unpack("!BBBB", self.connection.recv(4))
        assert version == SOCKS_VERSION

        if address_type == 1:  # IPv4
            address = socket.inet_ntoa(self.connection.recv(4))
        elif address_type == 3:  # Domain name
            domain_length = self.connection.recv(1)[0]
            address = self.connection.recv(domain_length)
            # address = socket.gethostbyname(address.decode("UTF-8"))  # 将域名转化为IP，这一行可以去掉
        elif address_type == 4:  # IPv6
            addr_ip = self.connection.recv(16)
            address = socket.inet_ntop(socket.AF_INET6, addr_ip)
        else:
            self.server.close_request(self.request)
            return
        port = struct.unpack('!H', self.connection.recv(2))[0]

        # 响应，只支持CONNECT请求
        try:
            if cmd == 1:  # CONNECT
                remote = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                remote.connect((address, port))
                bind_address = remote.getsockname()
                print('Connected to {} {}'.format(address, port))
            else:
                self.server.close_request(self.request)
            addr = struct.unpack("!I", socket.inet_aton(bind_address[0]))[0]
            port = bind_address[1]
            # reply = struct.pack("!BBBBIH", SOCKS_VERSION, 0, 0, address_type, addr, port)
            # 注意：按照标准协议，返回的应该是对应的address_type，但是实际测试发现，当address_type=3，也就是说是域名类型时，会出现卡死情况，但是将address_type该为1，则不管是IP类型和域名类型都能正常运行
            reply = struct.pack("!BBBBIH", SOCKS_VERSION, 0, 0, 1, addr, port)
        except Exception as err:
            print(err)
            # 响应拒绝连接的错误
            reply = generate_failed_reply(address_type, 5)
        self.connection.sendall(reply)

        # 建立连接成功，开始交换数据
        if reply[1] == 0 and cmd == 1:
            exchange_loop(self.connection, remote)
        self.server.close_request(self.request)

    def get_available_methods(self, n):
        methods = []
        for i in range(n):
            methods.append(ord(self.connection.recv(1)))
        return methods


if __name__ == '__main__':
    # 使用socketserver库的多线程服务器ThreadingTCPServer启动代理
    with ThreadingTCPServer(('127.0.0.1', 9011), SocksProxy) as server:
        server.serve_forever()

需要认证：

import logging
import select
import socket
import struct
from socketserver import StreamRequestHandler, ThreadingTCPServer

logging.basicConfig(level=logging.DEBUG)
SOCKS_VERSION = 5


def generate_failed_reply(address_type, error_number):
    return struct.pack("!BBBBIH", SOCKS_VERSION, error_number, 0, address_type, 0, 0)


def exchange_loop(client, remote):
    while True:
        # 等待数据
        r, w, e = select.select([client, remote], [], [])
        if client in r:
            data = client.recv(4096)
            if remote.send(data) <= 0:
                break
        if remote in r:
            data = remote.recv(4096)
            if client.send(data) <= 0:
                break


class SocksProxy(StreamRequestHandler):
    username = 'username'
    password = 'password'

    def handle(self):
        logging.info('Accepting connection from %s:%s' % self.client_address)
        
        # 协商
        # 从客户端读取并解包两个字节的数据
        header = self.connection.recv(2)
        version, nmethods = struct.unpack("!BB", header)
       
        # 设置socks5协议，METHODS字段的数目大于0
        assert version == SOCKS_VERSION
        assert nmethods > 0
        
        # 接受支持的方法
        methods = self.get_available_methods(nmethods)
        
        # 检查是否支持用户名/密码认证方式，不支持则断开连接
        if 2 not in set(methods):
            self.server.close_request(self.request)
            return
        
        # 发送协商响应数据包
        self.connection.sendall(struct.pack("!BB", SOCKS_VERSION, 2))
        
        # 校验用户名和密码
        if not self.verify_credentials():
            return
        
        # 请求
        version, cmd, _, address_type = struct.unpack("!BBBB", self.connection.recv(4))
        assert version == SOCKS_VERSION
        if address_type == 1:  # IPv4
            address = socket.inet_ntoa(self.connection.recv(4))
        elif address_type == 3:  # 域名
            domain_length = self.connection.recv(1)[0]
            address = self.connection.recv(domain_length)
        elif address_type == 4:  # IPv6
            addr_ip = self.connection.recv(16)
            address = socket.inet_ntop(socket.AF_INET6, addr_ip)
        else:
            self.server.close_request(self.request)
            return
        port = struct.unpack('!H', self.connection.recv(2))[0]
        
        # 响应，只支持CONNECT请求
        try:
            if cmd == 1:  # CONNECT
                remote = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                remote.connect((address, port))
                bind_address = remote.getsockname()
                logging.info('Connected to %s %s' % (address, port))
            else:
                self.server.close_request(self.request)
            addr = struct.unpack("!I", socket.inet_aton(bind_address[0]))[0]
            port = bind_address[1]
            reply = struct.pack("!BBBBIH", SOCKS_VERSION, 0, 0, 1, addr, port)
        except Exception as err:
            logging.error(err)
            # 响应拒绝连接的错误
            reply = generate_failed_reply(address_type, 5)
        self.connection.sendall(reply)
        
        # 建立连接成功，开始交换数据
        if reply[1] == 0 and cmd == 1:
            exchange_loop(self.connection, remote)
        self.server.close_request(self.request)

    def get_available_methods(self, n):
        methods = []
        for i in range(n):
            methods.append(ord(self.connection.recv(1)))
        return methods

    def verify_credentials(self):
        """校验用户名和密码"""
        version = ord(self.connection.recv(1))
        assert version == 1
        username_len = ord(self.connection.recv(1))
        username = self.connection.recv(username_len).decode('utf-8')
        password_len = ord(self.connection.recv(1))
        password = self.connection.recv(password_len).decode('utf-8')
        if username == self.username and password == self.password:
            # 验证成功, status = 0
            response = struct.pack("!BB", version, 0)
            self.connection.sendall(response)
            return True
        # 验证失败, status != 0
        response = struct.pack("!BB", version, 0xFF)
        self.connection.sendall(response)
        self.server.close_request(self.request)
        return False


if __name__ == '__main__':
    # 使用socketserver库的多线程服务器ThreadingTCPServer启动代理
    with ThreadingTCPServer(('127.0.0.1', 9011), SocksProxy) as server:
        server.serve_forever()

客户端

import socket
import socks
import requests
socks.set_default_proxy(socks.SOCKS5, "127.0.0.1", 9011, username=None, password=None)
socket.socket = socks.socksocket
print(requests.get('https://www.baidu.com').text)

参考：

https://zh.wikipedia.org/wiki/SOCKS

https://www.ietf.org/rfc/rfc1928.txt

https://rushter.com/blog/python-socks-server/

RFC 1928 - SOCKS5 协议中文文档

Python编写socks5服务器

socks5 协议详解