distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""SOCKS4a/5 client task — SOCKS proxy with SAM integration.
2
3Uses existing SOCKS5 parsing from socks_proxy.py. Adds SOCKS4a support
4and SAM-based I2P routing.
5
6Ported from net.i2p.i2ptunnel.socks.
7"""
8
9from __future__ import annotations
10
11import logging
12import socket
13import struct
14from dataclasses import dataclass
15
16from i2p_apps.i2ptunnel.config import TunnelDefinition
17from i2p_apps.i2ptunnel.forwarder import bridge
18from i2p_apps.i2ptunnel.socks_proxy import (
19 SOCKS_VERSION,
20 AUTH_NONE,
21 CMD_CONNECT,
22 ATYP_DOMAIN,
23 REPLY_SUCCESS,
24 REPLY_GENERAL_FAILURE,
25 REPLY_HOST_UNREACHABLE,
26 parse_socks_greeting,
27 build_socks_greeting_reply,
28 parse_socks_request,
29 build_socks_reply,
30)
31from i2p_apps.i2ptunnel.tasks import ClientTunnelTask
32
33logger = logging.getLogger(__name__)
34
35# SOCKS4 constants
36SOCKS4_VERSION = 4
37SOCKS4_CMD_CONNECT = 1
38SOCKS4_REPLY_GRANTED = 0x5A
39SOCKS4_REPLY_FAILED = 0x5B
40
41
42@dataclass
43class SOCKS4Request:
44 """Parsed SOCKS4/4a request."""
45 version: int
46 command: int
47 dest_port: int
48 dest_addr: str
49
50
51def parse_socks4a_request(data: bytes) -> SOCKS4Request:
52 """Parse a SOCKS4/4a CONNECT request.
53
54 Format: VER(1) CMD(1) DSTPORT(2) DSTIP(4) USERID(var,null-term) [DOMAIN(var,null-term)]
55
56 If DSTIP is 0.0.0.x (x != 0), it's SOCKS4a with domain after userid.
57 """
58 if len(data) < 8:
59 raise ValueError("SOCKS4 request too short")
60
61 version = data[0]
62 if version != SOCKS4_VERSION:
63 raise ValueError(f"Not SOCKS4: version={version}")
64
65 command = data[1]
66 port = struct.unpack("!H", data[2:4])[0]
67 ip_bytes = data[4:8]
68
69 # Find userid (null-terminated starting at byte 8)
70 userid_end = data.index(0x00, 8)
71
72 # Check if SOCKS4a (IP = 0.0.0.x where x != 0)
73 if ip_bytes[0] == 0 and ip_bytes[1] == 0 and ip_bytes[2] == 0 and ip_bytes[3] != 0:
74 # SOCKS4a: domain name follows userid null
75 domain_start = userid_end + 1
76 domain_end = data.index(0x00, domain_start)
77 domain = data[domain_start:domain_end].decode("utf-8")
78 return SOCKS4Request(version=version, command=command, dest_port=port, dest_addr=domain)
79 else:
80 # Regular SOCKS4: use the IP address
81 addr = socket.inet_ntoa(ip_bytes)
82 return SOCKS4Request(version=version, command=command, dest_port=port, dest_addr=addr)
83
84
85class SOCKSClientTask(ClientTunnelTask):
86 """SOCKS4a/5 proxy with SAM integration.
87
88 Handles both SOCKS5 (RFC 1928) and SOCKS4a protocols.
89 Routes .i2p destinations through SAM, non-.i2p to outproxy.
90 """
91
92 CMD_UDP_ASSOCIATE = 3
93
94 def __init__(self, config: TunnelDefinition, session) -> None:
95 super().__init__(config, session)
96 self._proxy_list = list(config.proxy_list)
97
98 @property
99 def _supports_udp(self) -> bool:
100 return False
101
102 @staticmethod
103 def _is_i2p(host: str) -> bool:
104 return host.endswith(".i2p")
105
106 async def _resolve(self, hostname: str) -> str | None:
107 if hostname.endswith(".b32.i2p"):
108 return hostname
109 if hostname.endswith(".i2p"):
110 return await self._session.lookup(hostname)
111 return hostname
112
113 async def handle_client(self, reader, writer) -> None:
114 try:
115 # Peek at first byte to determine SOCKS version
116 first_byte = await reader.read(1)
117 if not first_byte:
118 return
119
120 version = first_byte[0]
121
122 if version == SOCKS_VERSION:
123 await self._handle_socks5(first_byte, reader, writer)
124 elif version == SOCKS4_VERSION:
125 await self._handle_socks4(first_byte, reader, writer)
126 else:
127 writer.close()
128 except Exception:
129 logger.exception("Error in SOCKS proxy handler")
130
131 async def _handle_socks5(self, first_byte, reader, writer) -> None:
132 """Handle SOCKS5 connection."""
133 # Read rest of greeting
134 rest = await reader.read(256)
135 greeting_data = first_byte + rest
136 greeting = parse_socks_greeting(greeting_data)
137
138 if AUTH_NONE not in greeting.methods:
139 writer.write(build_socks_greeting_reply(0xFF))
140 await writer.drain()
141 return
142
143 writer.write(build_socks_greeting_reply(AUTH_NONE))
144 await writer.drain()
145
146 # Read request
147 request_data = await reader.read(262)
148 request = parse_socks_request(request_data)
149
150 # Check for UDP ASSOCIATE
151 if request.command == self.CMD_UDP_ASSOCIATE:
152 writer.write(build_socks_reply(0x07)) # Command not supported
153 await writer.drain()
154 return
155
156 if request.command != CMD_CONNECT:
157 writer.write(build_socks_reply(REPLY_GENERAL_FAILURE))
158 await writer.drain()
159 return
160
161 # Route
162 await self._route_socks5(request.dest_addr, request.dest_port, reader, writer)
163
164 async def _route_socks5(self, dest_addr, dest_port, reader, writer) -> None:
165 if self._is_i2p(dest_addr):
166 resolved = await self._resolve(dest_addr)
167 if resolved is None:
168 writer.write(build_socks_reply(REPLY_HOST_UNREACHABLE))
169 await writer.drain()
170 return
171
172 try:
173 remote_reader, remote_writer = await self._session.connect(resolved)
174 except Exception:
175 writer.write(build_socks_reply(REPLY_HOST_UNREACHABLE))
176 await writer.drain()
177 return
178
179 writer.write(build_socks_reply(REPLY_SUCCESS))
180 await writer.drain()
181 await bridge(reader, writer, remote_reader, remote_writer)
182 else:
183 # Non-I2P: forward to outproxy or reject
184 writer.write(build_socks_reply(REPLY_HOST_UNREACHABLE))
185 await writer.drain()
186
187 async def _handle_socks4(self, first_byte, reader, writer) -> None:
188 """Handle SOCKS4/4a connection."""
189 # Read enough for the request (variable length due to userid + domain)
190 rest = await reader.read(512)
191 data = first_byte + rest
192 request = parse_socks4a_request(data)
193
194 if request.command != SOCKS4_CMD_CONNECT:
195 writer.write(bytes([0x00, SOCKS4_REPLY_FAILED]) + b"\x00" * 6)
196 await writer.drain()
197 return
198
199 if self._is_i2p(request.dest_addr):
200 resolved = await self._resolve(request.dest_addr)
201 if resolved is None:
202 writer.write(bytes([0x00, SOCKS4_REPLY_FAILED]) + b"\x00" * 6)
203 await writer.drain()
204 return
205
206 try:
207 remote_reader, remote_writer = await self._session.connect(resolved)
208 except Exception:
209 writer.write(bytes([0x00, SOCKS4_REPLY_FAILED]) + b"\x00" * 6)
210 await writer.drain()
211 return
212
213 # Send SOCKS4 success reply
214 writer.write(bytes([0x00, SOCKS4_REPLY_GRANTED]) + b"\x00" * 6)
215 await writer.drain()
216 await bridge(reader, writer, remote_reader, remote_writer)
217 else:
218 writer.write(bytes([0x00, SOCKS4_REPLY_FAILED]) + b"\x00" * 6)
219 await writer.drain()