HTTP/3 및 SOCKS5를 통한 UDP

레지덴셜 프록시는 SOCKS5를 통해 UDP 트래픽을 전송합니다. 이를 사용하여 레지덴셜 IP를 통해 HTTP/3(QUIC)으로 웹사이트에 접속하거나, DNS, WebRTC, VoIP, 온라인 게임 등 기타 UDP 프로토콜을 라우팅할 수 있습니다.

작성 방법

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, 로그인 ID와 비밀번호를 설정한 후 실행하십시오. 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를 사용하려면, SOCKS5를 통해 UDP를 전달하는 도구를 경유하여 트래픽을 라우팅하십시오. 앱은 프록시를 인식하지 못하므로 평소와 같이 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 세션은 동일한 포트의 TCP 연결과 다른 출구 IP를 할당받을 수 있습니다.

자주 묻는 질문(FAQ)