HTTP/3 及通过 SOCKS5 传输的 UDP

住宅代理通过SOCKS5传输UDP流量。可利用它通过住宅IP地址访问基于HTTP/3（QUIC）的网站，并转发其他任何UDP协议：DNS、WebRTC、VoIP、网络游戏。

工作原理

HTTP/3 基于 QUIC 运行，而 QUIC 基于 UDP 运行。由于 HTTP 代理协议仅支持 TCP，因此 UDP 流量需通过 SOCKS5 传输。

在第一个数据报离开您的机器之前，会发生三次数据交换：

1. 连接与认证

   您的应用程序会向网关建立 SOCKS5 连接，并使用您列表中的用户名和密码进行登录。

2. 请求 UDP 中继

   它会发送一个UDP ASSOCIATE 请求（RFC 1928）。网关会返回一个中继地址：网关的 IP 地址以及 30000–39999 范围内的某个端口号。

3. 发送数据报

   该应用程序向该中继服务器发送 UDP 数据报。每个数据报开头都有一个简短的 SOCKS5 报头，其中指定了目标地址。流量从住宅出口 IP 发出，响应也通过同一路径返回。

连接设置

HTTP 和 SOCKS5 共享相同的主机、端口、登录名和密码——网关会自动识别协议类型。若要将连接字符串转换为 SOCKS5，只需更改协议方案，其他内容保持不变：

http://LOGIN:[email protected]:10000
socks5://LOGIN:[email protected]:10000

网关仅是入口点。无论您使用哪个网关，出口 IP 及其位置均来自列表设置——因此请选择距您服务器延迟最低的网关。

请自行核对

1. 检查连接

   任何 SOCKS5 客户端均可——curl 即可：

   curl -x socks5h://LOGIN:[email protected]:10000 https://api.ipify.org

   响应内容是代理的出口IP地址。使用socks5h://，域名由代理服务器解析，而非由您的计算机解析。

2. 检查 UDP

   该脚本仅使用 Python 标准库。它通过 SOCKS5 登录，打开一个 UDP 中继，并通过该中继发送一次 DNS 查询。OpenDNS 解析器返回myip.opendns.com 包含查询来源的地址，因此该脚本会输出您 UDP 流量的出口 IP 地址。

   将其保存为socks5_udp.py，设置您的登录名和密码，然后运行python socks5_udp.py.

   """UDP through a residential proxy: SOCKS5 login, UDP ASSOCIATE and one DNS query.
   
   The OpenDNS resolver answers "myip.opendns.com" with the IP the query came from,
   so the script prints the exit IP your UDP traffic leaves from.
   """
   import os
   import socket
   import struct
   
   PROXY_HOST = "res.proxy-seller.com"
   PROXY_PORT = 10000
   LOGIN = "LOGIN"
   PASSWORD = "PASSWORD"
   
   
   def recv_exact(sock, n):
       data = b""
       while len(data) < n:
           chunk = sock.recv(n - len(data))
           if not chunk:
               raise ConnectionError("proxy closed the connection")
           data += chunk
       return data
   
   
   def udp_associate(host, port, login, password):
       """SOCKS5 login (RFC 1929) and UDP ASSOCIATE (RFC 1928).
   
       Returns the TCP control connection and the UDP relay address.
       The relay works while the TCP connection stays open.
       """
       tcp = socket.create_connection((host, port), timeout=15)
       tcp.sendall(b"\x05\x01\x02")  # SOCKS5, one auth method: username/password
       if recv_exact(tcp, 2)[1] != 0x02:
           raise RuntimeError("username/password authentication is not available")
       tcp.sendall(b"\x01" + bytes([len(login)]) + login.encode()
                   + bytes([len(password)]) + password.encode())
       if recv_exact(tcp, 2)[1] != 0x00:
           raise RuntimeError("wrong login or password")
       tcp.sendall(b"\x05\x03\x00\x01" + bytes(6))  # UDP ASSOCIATE, client address 0.0.0.0:0
       _, reply, _, atyp = recv_exact(tcp, 4)
       if reply != 0x00:
           raise RuntimeError(f"UDP ASSOCIATE failed, reply code {reply}")
       if atyp == 0x01:
           relay_ip = socket.inet_ntoa(recv_exact(tcp, 4))
       elif atyp == 0x03:
           relay_ip = recv_exact(tcp, recv_exact(tcp, 1)[0]).decode()
       else:
           raise RuntimeError("IPv6 relay address is not handled in this example")
       relay_port = struct.unpack("!H", recv_exact(tcp, 2))[0]
       if relay_ip == "0.0.0.0":
           relay_ip = tcp.getpeername()[0]
       return tcp, (relay_ip, relay_port)
   
   
   def strip_udp_header(datagram):
       """Remove the SOCKS5 UDP header from a datagram received from the relay."""
       atyp = datagram[3]
       if atyp == 0x01:  # IPv4
           return datagram[10:]
       if atyp == 0x04:  # IPv6
           return datagram[22:]
       return datagram[7 + datagram[4]:]  # domain name
   
   
   if __name__ == "__main__":
       tcp, relay = udp_associate(PROXY_HOST, PROXY_PORT, LOGIN, PASSWORD)
       print("UDP relay:", f"{relay[0]}:{relay[1]}")
   
       # DNS query "myip.opendns.com, type A" without EDNS: the answer ends with the IPv4 address
       query = (os.urandom(2) + b"\x01\x00\x00\x01\x00\x00\x00\x00\x00\x00"
                + b"\x04myip\x07opendns\x03com\x00" + b"\x00\x01\x00\x01")
       # every datagram to the relay starts with a SOCKS5 UDP header: RSV, FRAG, ATYP=IPv4, destination
       header = b"\x00\x00\x00\x01" + socket.inet_aton("208.67.222.222") + struct.pack("!H", 53)
   
       udp = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
       udp.settimeout(10)
       udp.sendto(header + query, relay)
       answer = strip_udp_header(udp.recv(4096))
       print("Exit IP over UDP:", socket.inet_ntoa(answer[-4:]))
       udp.close()
       tcp.close()

   输出示例：

   UDP relay: 103.152.136.28:39631
   Exit IP over UDP: 203.0.113.45

3. 发送一个 HTTP/3 请求

   对于 HTTP/3，请求的 QUIC 数据包会通过相同的 UDP 中继传输。此示例使用的是aioquic 库（Python 3.11 或更高版本）。它加载https://cloudflare-quic.com/cdn-cgi/trace，一个显示协议和客户端 IP 的纯文本页面。

   安装依赖项，将脚本保存为http3_via_socks5.py 紧邻socks5_udp.py，并运行它：

   pip install aioquic certifi
   python http3_via_socks5.py

   """HTTP/3 (QUIC) request to a website through a residential proxy.
   
   QUIC packets go to the SOCKS5 UDP relay; the website sees an HTTP/3 connection from the exit IP.
   Requires: pip install aioquic certifi, and socks5_udp.py from the previous example in the same folder.
   """
   import socket
   import struct
   import time
   
   import certifi
   from aioquic.h3.connection import H3_ALPN, H3Connection
   from aioquic.h3.events import DataReceived, HeadersReceived
   from aioquic.quic.configuration import QuicConfiguration
   from aioquic.quic.connection import QuicConnection
   from aioquic.quic.events import ConnectionTerminated, HandshakeCompleted
   
   from socks5_udp import LOGIN, PASSWORD, PROXY_HOST, PROXY_PORT, strip_udp_header, udp_associate
   
   TARGET_HOST = "cloudflare-quic.com"
   TARGET_PATH = "/cdn-cgi/trace"  # plain text: shows the protocol (http=http/3) and the client IP
   
   tcp, relay = udp_associate(PROXY_HOST, PROXY_PORT, LOGIN, PASSWORD)
   udp = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
   # the destination goes into the SOCKS5 UDP header as a domain name (ATYP=3): the proxy resolves it
   header = b"\x00\x00\x00\x03" + bytes([len(TARGET_HOST)]) + TARGET_HOST.encode() + struct.pack("!H", 443)
   
   config = QuicConfiguration(is_client=True, alpn_protocols=H3_ALPN, server_name=TARGET_HOST)
   config.load_verify_locations(certifi.where())
   quic = QuicConnection(configuration=config)
   peer = ("0.0.0.0", 443)  # placeholder: only the proxy knows the real address
   quic.connect(peer, now=time.time())
   h3 = H3Connection(quic)
   
   
   def send_pending():
       for datagram, _ in quic.datagrams_to_send(now=time.time()):
           udp.sendto(header + datagram, relay)
   
   
   send_pending()
   stream_id, body, done = None, b"", False
   deadline = time.time() + 20
   while not done and time.time() < deadline:
       timer = quic.get_timer()
       udp.settimeout(0.5 if timer is None else min(0.5, max(0.01, timer - time.time())))
       try:
           quic.receive_datagram(strip_udp_header(udp.recv(65535)), peer, now=time.time())
       except socket.timeout:
           if timer is not None and time.time() >= timer:
               quic.handle_timer(now=time.time())
       event = quic.next_event()
       while event is not None:
           if isinstance(event, HandshakeCompleted):
               stream_id = quic.get_next_available_stream_id()
               h3.send_headers(stream_id, [
                   (b":method", b"GET"), (b":scheme", b"https"),
                   (b":authority", TARGET_HOST.encode()), (b":path", TARGET_PATH.encode()),
               ], end_stream=True)
           elif isinstance(event, ConnectionTerminated):
               done = True
           for h3_event in h3.handle_event(event):
               if isinstance(h3_event, (HeadersReceived, DataReceived)) and h3_event.stream_id == stream_id:
                   if isinstance(h3_event, DataReceived):
                       body += h3_event.data
                   done = done or h3_event.stream_ended
           event = quic.next_event()
       send_pending()
   
   print(body.decode() or "No HTTP/3 response within 20 seconds")
   quic.close()
   send_pending()
   udp.close()
   tcp.close()

   输出（精简版）：

   ip=203.0.113.45
   http=http/3
   loc=GB
   tls=TLSv1.3

   http=http/3 表示该网站通过 HTTP/3 接收了该请求。ip 以及loc 是代理服务器的出口IP及其所在国家。

浏览器及其他应用中的 HTTP/3

只有当应用程序支持通过 SOCKS5 传输 UDP 时，才能通过代理发送 UDP（UDP ASSOCIATE)。

要在浏览器或其他应用中使用 HTTP/3，需将流量通过一个能够将 UDP 数据包转发至 SOCKS5 的工具进行路由。应用完全不会察觉到代理的存在，因此仍像往常一样使用 HTTP/3，而该工具则会通过网关发送 QUIC 数据包。

{
  "type": "socks",
  "tag": "residential",
  "server": "res.proxy-seller.com",
  "server_port": 10000,
  "version": "5",
  "username": "LOGIN",
  "password": "PASSWORD"
}

限制条件

• UDP 仅通过 SOCKS5 工作。HTTP 代理连接采用 TCP 协议。

• 不支持通过 HTTPS（TLS）或 HTTP/3 连接到代理服务器本身——请使用http:// 或socks5://.

• UDP 会话在同一端口上获得的出口 IP 可能与 TCP 连接不同。

常见问题解答