distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for I2P tunnel infrastructure — HTTP proxy, SOCKS proxy, server tunnels."""
2
3import struct
4import pytest
5
6from i2p_apps.i2ptunnel.http_proxy import (
7 HTTPRequest,
8 HTTPProxy,
9 parse_http_request,
10 extract_i2p_destination,
11)
12from i2p_apps.i2ptunnel.socks_proxy import (
13 SOCKS_VERSION,
14 AUTH_NONE,
15 CMD_CONNECT,
16 ATYP_DOMAIN,
17 REPLY_SUCCESS,
18 REPLY_GENERAL_FAILURE,
19 REPLY_HOST_UNREACHABLE,
20 SOCKSGreeting,
21 SOCKSRequest,
22 SOCKSProxy,
23 parse_socks_greeting,
24 build_socks_greeting_reply,
25 parse_socks_request,
26 build_socks_reply,
27)
28from i2p_apps.i2ptunnel.server_tunnel import ServerTunnel
29from i2p_apps.i2ptunnel.tunnel_manager import I2PTunnelManager
30
31
32# ---------------------------------------------------------------------------
33# HTTPProxy tests
34# ---------------------------------------------------------------------------
35
36class TestParseHTTPRequest:
37 """Test HTTP request first-line parsing."""
38
39 def test_connect_request(self):
40 req = parse_http_request("CONNECT example.i2p:443 HTTP/1.1")
41 assert req.method == "CONNECT"
42 assert req.host == "example.i2p"
43 assert req.port == 443
44 assert req.is_connect is True
45 assert req.raw_first_line == "CONNECT example.i2p:443 HTTP/1.1"
46
47 def test_regular_get_request(self):
48 req = parse_http_request("GET http://forum.i2p/index.html HTTP/1.1")
49 assert req.method == "GET"
50 assert req.host == "forum.i2p"
51 assert req.port == 80
52 assert req.path == "/index.html"
53 assert req.is_connect is False
54
55 def test_regular_get_with_port(self):
56 req = parse_http_request("GET http://forum.i2p:8080/api HTTP/1.1")
57 assert req.host == "forum.i2p"
58 assert req.port == 8080
59 assert req.path == "/api"
60
61 def test_connect_without_port(self):
62 req = parse_http_request("CONNECT example.i2p HTTP/1.1")
63 assert req.host == "example.i2p"
64 assert req.port == 443 # default for CONNECT
65
66 def test_post_request(self):
67 req = parse_http_request("POST http://mail.i2p/send HTTP/1.1")
68 assert req.method == "POST"
69 assert req.host == "mail.i2p"
70 assert req.path == "/send"
71 assert req.is_connect is False
72
73
74class TestExtractI2PDestination:
75 """Test .i2p destination extraction."""
76
77 def test_plain_i2p_host(self):
78 assert extract_i2p_destination("forum.i2p") == "forum.i2p"
79
80 def test_subdomain_i2p(self):
81 assert extract_i2p_destination("www.forum.i2p") == "www.forum.i2p"
82
83 def test_non_i2p_host(self):
84 assert extract_i2p_destination("example.com") is None
85
86 def test_empty_host(self):
87 assert extract_i2p_destination("") is None
88
89 def test_i2p_in_middle(self):
90 # "i2p.example.com" is NOT an i2p destination
91 assert extract_i2p_destination("i2p.example.com") is None
92
93
94class TestHTTPProxy:
95 """Test HTTPProxy instance behavior."""
96
97 def test_default_listen_address(self):
98 proxy = HTTPProxy()
99 assert proxy.listen_address == ("127.0.0.1", 4444)
100
101 def test_custom_listen_address(self):
102 proxy = HTTPProxy(listen_host="0.0.0.0", listen_port=8888)
103 assert proxy.listen_address == ("0.0.0.0", 8888)
104
105 def test_not_running_initially(self):
106 proxy = HTTPProxy()
107 assert proxy.is_running is False
108
109 def test_build_502_response(self):
110 proxy = HTTPProxy()
111 resp = proxy.build_error_response(502, "Bad Gateway")
112 assert b"502" in resp
113 assert b"Bad Gateway" in resp
114 assert resp.startswith(b"HTTP/1.1 502")
115
116
117# ---------------------------------------------------------------------------
118# SOCKSProxy tests
119# ---------------------------------------------------------------------------
120
121class TestParseSOCKSGreeting:
122 """Test SOCKS5 greeting parsing."""
123
124 def test_no_auth_greeting(self):
125 # version=5, nmethods=1, methods=[0 (no auth)]
126 data = bytes([0x05, 0x01, 0x00])
127 greeting = parse_socks_greeting(data)
128 assert greeting.version == 5
129 assert greeting.methods == [AUTH_NONE]
130
131 def test_multi_method_greeting(self):
132 # version=5, nmethods=2, methods=[0, 2]
133 data = bytes([0x05, 0x02, 0x00, 0x02])
134 greeting = parse_socks_greeting(data)
135 assert greeting.version == 5
136 assert greeting.methods == [0, 2]
137
138 def test_invalid_version_raises(self):
139 data = bytes([0x04, 0x01, 0x00]) # SOCKS4
140 with pytest.raises(ValueError, match="[Vv]ersion"):
141 parse_socks_greeting(data)
142
143
144class TestBuildSOCKSGreetingReply:
145 """Test SOCKS5 greeting reply construction."""
146
147 def test_no_auth_reply(self):
148 reply = build_socks_greeting_reply(AUTH_NONE)
149 assert reply == bytes([0x05, 0x00])
150
151 def test_custom_method_reply(self):
152 reply = build_socks_greeting_reply(0x02)
153 assert reply == bytes([0x05, 0x02])
154
155
156class TestParseSOCKSRequest:
157 """Test SOCKS5 CONNECT request parsing."""
158
159 def test_domain_connect(self):
160 # VER=5, CMD=1 (connect), RSV=0, ATYP=3 (domain)
161 domain = b"forum.i2p"
162 data = bytes([0x05, 0x01, 0x00, 0x03, len(domain)]) + domain + struct.pack("!H", 80)
163 req = parse_socks_request(data)
164 assert req.version == 5
165 assert req.command == CMD_CONNECT
166 assert req.address_type == ATYP_DOMAIN
167 assert req.dest_addr == "forum.i2p"
168 assert req.dest_port == 80
169
170 def test_domain_with_high_port(self):
171 domain = b"mail.i2p"
172 data = bytes([0x05, 0x01, 0x00, 0x03, len(domain)]) + domain + struct.pack("!H", 8443)
173 req = parse_socks_request(data)
174 assert req.dest_addr == "mail.i2p"
175 assert req.dest_port == 8443
176
177 def test_invalid_version_raises(self):
178 data = bytes([0x04, 0x01, 0x00, 0x03, 0x05]) + b"a.i2p" + struct.pack("!H", 80)
179 with pytest.raises(ValueError, match="[Vv]ersion"):
180 parse_socks_request(data)
181
182
183class TestBuildSOCKSReply:
184 """Test SOCKS5 reply construction."""
185
186 def test_success_reply(self):
187 reply = build_socks_reply(REPLY_SUCCESS)
188 assert reply[0] == 0x05 # version
189 assert reply[1] == REPLY_SUCCESS
190 assert reply[2] == 0x00 # reserved
191 assert len(reply) == 10 # VER + REP + RSV + ATYP(1) + ADDR(4) + PORT(2)
192
193 def test_failure_reply(self):
194 reply = build_socks_reply(REPLY_GENERAL_FAILURE)
195 assert reply[1] == REPLY_GENERAL_FAILURE
196
197 def test_host_unreachable_reply(self):
198 reply = build_socks_reply(REPLY_HOST_UNREACHABLE)
199 assert reply[1] == REPLY_HOST_UNREACHABLE
200
201 def test_custom_bind(self):
202 reply = build_socks_reply(REPLY_SUCCESS, bind_addr="127.0.0.1", bind_port=1080)
203 port_bytes = reply[-2:]
204 port = struct.unpack("!H", port_bytes)[0]
205 assert port == 1080
206
207
208class TestSOCKSProxy:
209 """Test SOCKSProxy instance."""
210
211 def test_default_listen_address(self):
212 proxy = SOCKSProxy()
213 assert proxy.listen_address == ("127.0.0.1", 4445)
214
215 def test_not_running_initially(self):
216 proxy = SOCKSProxy()
217 assert proxy.is_running is False
218
219
220# ---------------------------------------------------------------------------
221# ServerTunnel tests
222# ---------------------------------------------------------------------------
223
224class TestServerTunnel:
225 """Test server tunnel dataclass."""
226
227 def test_create_with_local_address(self):
228 tunnel = ServerTunnel(name="web", local_host="127.0.0.1", local_port=8080)
229 assert tunnel.name == "web"
230 assert tunnel.local_host == "127.0.0.1"
231 assert tunnel.local_port == 8080
232 assert tunnel.i2p_port == 0
233 assert tunnel.running is False
234
235 def test_create_with_custom_i2p_port(self):
236 tunnel = ServerTunnel(name="irc", local_host="127.0.0.1", local_port=6667, i2p_port=6668)
237 assert tunnel.i2p_port == 6668
238
239 def test_forward_data_mock(self):
240 """ServerTunnel should track forwarded byte counts."""
241 tunnel = ServerTunnel(name="web", local_host="127.0.0.1", local_port=8080)
242 assert tunnel.bytes_forwarded == 0
243 tunnel.record_forwarded(1024)
244 assert tunnel.bytes_forwarded == 1024
245 tunnel.record_forwarded(512)
246 assert tunnel.bytes_forwarded == 1536
247
248
249# ---------------------------------------------------------------------------
250# I2PTunnelManager tests
251# ---------------------------------------------------------------------------
252
253class TestI2PTunnelManager:
254 """Test tunnel manager add/remove/list."""
255
256 def test_add_and_list(self):
257 mgr = I2PTunnelManager()
258 proxy = HTTPProxy()
259 mgr.add_tunnel("http", proxy)
260 assert mgr.list_tunnels() == ["http"]
261 assert mgr.tunnel_count == 1
262
263 def test_remove_tunnel(self):
264 mgr = I2PTunnelManager()
265 proxy = HTTPProxy()
266 mgr.add_tunnel("http", proxy)
267 assert mgr.remove_tunnel("http") is True
268 assert mgr.tunnel_count == 0
269
270 def test_remove_nonexistent(self):
271 mgr = I2PTunnelManager()
272 assert mgr.remove_tunnel("nope") is False
273
274 def test_get_tunnel_by_name(self):
275 mgr = I2PTunnelManager()
276 proxy = SOCKSProxy()
277 mgr.add_tunnel("socks", proxy)
278 assert mgr.get_tunnel("socks") is proxy
279
280 def test_get_tunnel_missing(self):
281 mgr = I2PTunnelManager()
282 assert mgr.get_tunnel("missing") is None
283
284 def test_add_duplicate_raises(self):
285 mgr = I2PTunnelManager()
286 mgr.add_tunnel("http", HTTPProxy())
287 with pytest.raises(ValueError, match="already exists"):
288 mgr.add_tunnel("http", HTTPProxy())
289
290 def test_mixed_tunnel_types(self):
291 mgr = I2PTunnelManager()
292 mgr.add_tunnel("http", HTTPProxy())
293 mgr.add_tunnel("socks", SOCKSProxy())
294 mgr.add_tunnel("server", ServerTunnel(name="web", local_host="127.0.0.1", local_port=80))
295 assert mgr.tunnel_count == 3
296 assert sorted(mgr.list_tunnels()) == ["http", "server", "socks"]