或许你没听说过socks5，但你一定听说过SS，SS内部使用的正是socks5协议。

 

什么是socks5？

socks是”SocketS”的缩写，因此socks5也叫sockets5。

socks是一种网络传输协议，根据OSI七层模型来划分，SOCKS属于会话层协议，位于表示层与传输层之间。

当防火墙后的客户端要访问外部的服务器时，就跟socks代理服务器连接。该协议设计之初是为了让有权限的用户可以穿过过防火墙的限制，使得高权限用户可以访问外部资源。经过10余年的时间，大量的网络应用程序都支持socks5代理。

这个协议最初由David Koblas开发，而后由NEC的Ying-Da Lee将其扩展到版本4，最新协议是版本5，与前一版本相比，socks5做了以下增强：

• 增加对UDP协议的支持；
• 支持多种用户身份验证方式和通信加密方式；
• 修改了socks服务器进行域名解析的方法，使其更加优雅；

 

socks5使用场景

socks协议的设计初衷是在保证网络隔离的情况下，提高部分人员的网络访问权限，但人们现在主要用来突破网络通信限制，这和该协议的设计初衷正好相反。

下面是比较常见的使用场景：

• 美国某网游的服务器仅允许本国的IP进行连接。非美国玩家为了突破这种限制，可以找一个该地区的socks5代理服务器，然后用PSD接管网游客户端，通过socks5代理服务器连接游戏服务器。这样游戏服务器就会认为该玩家的客户端位于本地区，从而允许该玩家进行游戏（在天朝也叫科学**，属于正向代理）。
• 某服务器的防火墙仅允许部分端口（如http的80端口）通信，那么可以利用socks5协议和一个打开80端口监听的socks5服务器连接，从而可以连接公网上其他端口的服务器。利用一些额外的技术手段，甚至可以骗过内部的http代理服务器，这时在使用内网http代理上网的环境下也可以不受限制的使用网络服务，这称之为socks over HTTP（我们常说的穿墙）。
• 内网穿透：在大学里，学校给我们提供了很多服务器资源，我们可以在内网使用。但放寒假回家后，无法进入学校内网，也就无法连接上内网的服务器资源。解决办法：在公网的VPS上搭一个socks代理，并将内网的一台web服务器和该VPS的socks端口打通，通过这台web服务器便可以访问所有内网服务器资源（也就是反向代理，常见的花生壳nat穿透和这个类似）。

当然，使用代理服务器后，将不可避免的出现通信延迟，所以应该尽量选择同网络（同运营商）、距离近的服务器。

 

与HTTP代理的对比

socks支持多种用户身份验证方式和通信加密方式。

socks工作在比HTTP代理更低的网络层：socks使用握手协议来通知代理软件其客户端试图进行的连接socks，然后尽可能透明地进行操作，而常规代理可能会解释和重写报头（例如，使用另一种底层协议，例如FTP；然而，HTTP代理只是将HTTP请求转发到所需的HTTP服务器）。

socks5代理支持转发UDP报文，而HTTP属于tcp协议，不支持UDP报文的转发。

虽然HTTP代理有不同的使用模式，CONNECT方法允许转发TCP连接；然而，socks代理还可以转发UDP流量和反向代理，而HTTP代理不能。HTTP代理更适合HTTP协议，执行更高层次的过滤；socks不管应用层是什么协议，只要是传输层是TCP/UDP协议就可以代理。

 

socks5协议详解

socks5连接协议

创建与socks5服务器的TCP连接后，客户端需要先发送请求来协商版本及认证方式，格式为：

• VER：socks版本（在socks5中是0x05）；
• NMETHODS：客户端支持的认证机制数目；
• METHODS：客户端支持的认证方式列表，每个方法占1字节。

服务器从客户端提供的方法中选择一个最优的方法并通过以下消息通知客户端（贪心算法：双方都支持、安全性最高）：

• VER：socks版本（在socks5中是0x05）；
• METHOD：服务端选中的方法（若返回0xFF表示没有方法被选中，客户端需要关闭连接）；

METHOD字段的值可以取如下值：

• 0x00  NO AUTHENTICATION REQUIRED
• 0x01  GSSAPI
• 0x02  USERNAME/PASSWORD
• 0x03  to X’7F’ IANA ASSIGNED
• 0x80  to X’FE’ RESERVED FOR PRIVATE METHODS
• 0xFF  NO ACCEPTABLE METHODS

之后客户端和服务端根据选定的认证方式执行对应的认证。认证结束后客户端就可以发送请求信息（如果认证方法有特殊封装要求，请求必须按照方法所定义的方式进行封装）。

socks5请求格式：

• VER：socks版本（在socks5中是0x05）
• CMD：SOCK的命令码：
  • CONNECT 0x01
  • BIND 0x02
  • UDP ASSOCIATE 0x03

CONNECT 用于 TCP 场景，BIND 主要用于 FTP 等罕见场景，UDP ASSOCIATE 用于 UDP 场景。

• RSV：保留字段，固定取值 0x00
• ATYP：地址类型：
  • IP V4地址: 0x01
  • 域名地址: 0x03
  • IP V6地址: 0x04

• DST.ADDR：目的地址

• DST.PORT：目的端口

服务器按以下格式回应客户端的请求：

• VER：socks版本（在socks5中是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~0xFF  unassigned
• RSV：保留字段，默认设置为0x00
• ATYP：地址类型：
  • IP V4 address: 0x01
  • DOMAINNAME: 0x03
  • IP V6 address: 0x04
• BND.ADDR：服务器绑定的地址
• BND.PORT：服务器绑定的端口

如果被选中的方法包括有认证信息的封装、完整性和/或机密性相关检查，则server端在发送响应包时也需要把这些响应消息封装进去。

至此，客户端和服务端的连接已经建立完成，后续客户端可以通过 SOCKS5 服务端与目的地址进行数据传输。

 

socks5认证协议

在客户端、服务端协商好使用用户名密码认证后（METHOD字段为0x02  USERNAME/PASSWORD），客户端发出用户名密码，格式为：

• VER：鉴定协议版本
• ULEN：用户名长度
• UNAME：用户名
• PLEN：密码长度
• PASSWD：密码

服务器鉴定后发出如下回应：

• VER：鉴定协议版本
• STATUS：鉴定状态

其中鉴定状态 0x00 表示成功，0x01 表示失败。

 

python实现socks5代理

!/usr/bin/env python
coding:utf-8

import os
import select
import socket
import struct
from socketserver import ThreadingMixIn, TCPServer, StreamRequestHandler
import datetime

SOCKS_VERSION = 5
nextConn = []

class ThreadingTCPServer(ThreadingMixIn, TCPServer):
    pass

class SocksProxy(StreamRequestHandler):
    whiteList = ["github.com","gitee.com"]
    time = 0

    def handle(self):
        print('Accepting connection from:')
        print(self.client_address)
    
        # greeting header 
        # read and unpack 2 bytes from a client
        header = self.connection.recv(2) 
        version, nmethods = struct.unpack("!BB", header) 

        # socks 5 
        assert version == SOCKS_VERSION 
        assert nmethods > 0 

        # get available methods 
        methods = self.get_available_methods(nmethods) 

        # send welcome message 
        self.connection.sendall(struct.pack("!BB", SOCKS_VERSION, 0)) 

        # request 
        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 = ord(self.connection.recv(1)) 
            address = self.connection.recv(domain_length) 

        print("address is:") 
        print(address) 

        port = struct.unpack('!H', self.connection.recv(2))[0] 

        # reply 
        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 %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, address_type, addr, port)

        except Exception as err: 
            print(err) 
            # return connection refused error 
            reply = self.generate_failed_reply(address_type, 5) 

        self.connection.sendall(reply) 

        # establish data exchange 
        if reply[1] == 0 and cmd == 1: 
            self.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 generate_failed_reply(self, address_type, error_number): 
        return struct.pack("!BBBBIH", SOCKS_VERSION, error_number, 0, address_type, 0, 0)

    #判断连接方式、目标网站等等
    #基于http跳转https传输
    #git可以clone，禁止push
    def checkConn(self,msg): 
        try: 
            header = bytes.decode(msg) 
            print(header) 
            headerList = header.split("\r\n") 
         
            if headerList[0][:3] != "GET": 
                return False 
            if "git-receive-pack" in headerList[0].split("/")[-2]: 
                return False 
            if headerList[1].split(":")[-1].strip() not in self.whiteList: 
                return False 
            if headerList[2].split(":")[1].strip()[:3] != "git": 
                return False 
            return True 
        except: 
            return False

    # http跳转是第一遍连接http，然后中断连接，重新连接https 
    # 所以成功连接http时记录IP，成功连接上https时立即删除IP，以防同一IP多重连接 
    # 不在记录中的https连接全部break
    def exchange_loop(self, client, remote): 
        while True: 
            # wait until client or remote is available for read 
            r, w, e = select.select([client, remote], [], []) 
            if client in r: 
                try: 
                    data = client.recv(4096) 
                except: 
                    break 
                if self.time == 0: 
                    #if is check ok,note ip 
                    if self.checkConn(data): 
                        nextConn.append(self.client_address[0]) 
                        if remote.send(data) <= 0: 
                            break 

                    #if ip noted,remove ip for multi attach 
                    elif self.client_address[0] in nextConn: 
                        if self.client_address[0] in nextConn: 
                            nextConn.remove(self.client_address[0]) 
                        if remote.send(data) <= 0: 
                            break 
                    else: 
                        break 
                else: 
                    if remote.send(data) <= 0: 
                        break 
                self.time += 1 

            if remote in r: 
                try: 
                    data = remote.recv(4096) 
                except: 
                    break 
                if client.send(data) <= 0: 
                    break

if __name__ == '__main__':
     IP = "127.0.0.1"
     with ThreadingTCPServer((IP, 1080), SocksProxy) as server:
         server.serve_forever()

那么这个socks代理有什么用呢？

比如，前面文章所说的基于socks5的git的http协议代理，同理pip等类似工具也可使用，在封闭内网中通过搭建此服务的代理服务器中转连接，clone下载外网第三方仓库或扩展等，加上使用日志、流量统计、守护进程以及邮件报警等功能即可用于生产开发环境。

目前的问题是，对于http跳转https的放行判断，仍不满意，却没有更好的方案，只能暂时这样，以后有了什么想法再改吧。

本文转载自知寒，并根据自己的理解稍作修改，可能有谬误之处，仅作记录。

python代码是在rushter的基础上修改而来，用于非账号密码连接因此去掉了用户认证部分。

---

尊重不一定是接受。

——伏尔泰