distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for HTTPClientTask — header filtering, outproxy, address helpers."""
2
3from unittest.mock import AsyncMock, MagicMock
4
5import pytest
6
7from i2p_apps.i2ptunnel.config import TunnelDefinition, TunnelType
8from i2p_apps.i2ptunnel.http_client import HTTPClientTask, _STRIPPED_OUTBOUND, _I2P_USER_AGENT
9
10
11def _make_task(proxy_list=None):
12 config = TunnelDefinition(
13 name="test-http",
14 type=TunnelType.HTTPCLIENT,
15 listen_port=0,
16 interface="127.0.0.1",
17 proxy_list=proxy_list or [],
18 )
19 session = MagicMock()
20 session.lookup = AsyncMock(return_value=None)
21 session.connect = AsyncMock()
22 return HTTPClientTask(config, session)
23
24
25class TestHeaderFiltering:
26 def test_strips_privacy_headers(self):
27 task = _make_task()
28 headers = {
29 "Referer": "http://example.com",
30 "Via": "proxy",
31 "X-Forwarded-For": "1.2.3.4",
32 "From": "user@example.com",
33 "Host": "site.i2p",
34 "Accept": "text/html",
35 }
36 filtered = task._filter_outbound_headers(headers, is_i2p=True)
37 assert "Referer" not in filtered
38 assert "Via" not in filtered
39 assert "X-Forwarded-For" not in filtered
40 assert "From" not in filtered
41 assert "Host" in filtered
42 assert "Accept" in filtered
43
44 def test_replaces_user_agent_for_i2p(self):
45 task = _make_task()
46 headers = {"User-Agent": "Mozilla/5.0"}
47 filtered = task._filter_outbound_headers(headers, is_i2p=True)
48 assert filtered["User-Agent"] == _I2P_USER_AGENT
49
50 def test_keeps_user_agent_for_clearnet(self):
51 task = _make_task()
52 headers = {"User-Agent": "Mozilla/5.0"}
53 filtered = task._filter_outbound_headers(headers, is_i2p=False)
54 assert filtered["User-Agent"] == "Mozilla/5.0"
55
56 def test_proxy_headers_stripped(self):
57 task = _make_task()
58 headers = {
59 "Proxy-Connection": "keep-alive",
60 "Proxy-Authorization": "Basic xxx",
61 }
62 filtered = task._filter_outbound_headers(headers, is_i2p=True)
63 assert len(filtered) == 0
64
65
66class TestAddressHelpers:
67 def test_extract_address_helper(self):
68 task = _make_task()
69 host, helper = task._extract_address_helper(
70 "http://site.i2p/page?i2paddresshelper=AAAA"
71 )
72 assert host == "site.i2p"
73 assert helper == "AAAA"
74
75 def test_no_address_helper(self):
76 task = _make_task()
77 host, helper = task._extract_address_helper("http://site.i2p/page")
78 assert host == "site.i2p"
79 assert helper is None
80
81 def test_cache_and_retrieve(self):
82 task = _make_task()
83 task._cache_address_helper("site.i2p", "DEST_BASE64")
84 assert task._get_cached_helper("site.i2p") == "DEST_BASE64"
85 assert task._get_cached_helper("other.i2p") is None
86
87
88class TestOutproxy:
89 def test_has_outproxy(self):
90 task = _make_task(proxy_list=["proxy.i2p"])
91 assert task._has_outproxy()
92
93 def test_no_outproxy(self):
94 task = _make_task(proxy_list=[])
95 assert not task._has_outproxy()
96
97 def test_pick_outproxy(self):
98 task = _make_task(proxy_list=["proxy1.i2p", "proxy2.i2p"])
99 choice = task._pick_outproxy()
100 assert choice in ("proxy1.i2p", "proxy2.i2p")
101
102 def test_pick_outproxy_avoids_failed(self):
103 task = _make_task(proxy_list=["proxy1.i2p", "proxy2.i2p"])
104 task._failed_outproxies.add("proxy1.i2p")
105 choice = task._pick_outproxy()
106 assert choice == "proxy2.i2p"
107
108 def test_pick_outproxy_resets_when_all_failed(self):
109 task = _make_task(proxy_list=["proxy1.i2p"])
110 task._failed_outproxies.add("proxy1.i2p")
111 choice = task._pick_outproxy()
112 assert choice == "proxy1.i2p"
113 assert len(task._failed_outproxies) == 0
114
115 def test_pick_outproxy_empty_list(self):
116 task = _make_task(proxy_list=[])
117 assert task._pick_outproxy() == ""
118
119
120class TestRequestClassification:
121 def test_is_i2p_request(self):
122 assert HTTPClientTask._is_i2p_request("site.i2p")
123 assert HTTPClientTask._is_i2p_request("a.b32.i2p")
124 assert not HTTPClientTask._is_i2p_request("example.com")
125
126 def test_is_localhost(self):
127 assert HTTPClientTask._is_localhost("127.0.0.1")
128 assert HTTPClientTask._is_localhost("localhost")
129 assert HTTPClientTask._is_localhost("::1")
130 assert HTTPClientTask._is_localhost("0.0.0.0")
131 assert not HTTPClientTask._is_localhost("example.com")
132
133
134class TestErrorPages:
135 def test_error_page_contains_message(self):
136 task = _make_task()
137 page = task._error_page("dnf", 503)
138 assert b"503" in page
139 assert b"Destination Not Found" in page
140 assert b"text/html" in page
141
142 def test_error_page_unknown_type(self):
143 task = _make_task()
144 page = task._error_page("unknown_type", 500)
145 assert b"500" in page
146 assert b"Error" in page
147
148
149class TestRequestRebuilding:
150 def test_rebuild_get(self):
151 result = HTTPClientTask._rebuild_request(
152 "GET", "/path", {"Host": "site.i2p"}, b""
153 )
154 assert b"GET /path HTTP/1.1" in result
155 assert b"Host: site.i2p" in result
156
157 def test_rebuild_post_with_body(self):
158 body = b"key=value"
159 result = HTTPClientTask._rebuild_request(
160 "POST", "/submit", {"Content-Length": "9"}, body
161 )
162 assert b"POST /submit HTTP/1.1" in result
163 assert result.endswith(body)
164
165
166class TestResolve:
167 @pytest.mark.asyncio
168 async def test_b32_passthrough(self):
169 task = _make_task()
170 result = await task._resolve("abcdef.b32.i2p")
171 assert result == "abcdef.b32.i2p"
172
173 @pytest.mark.asyncio
174 async def test_cached_helper(self):
175 task = _make_task()
176 task._cache_address_helper("site.i2p", "CACHED_DEST")
177 result = await task._resolve("site.i2p")
178 assert result == "CACHED_DEST"
179
180 @pytest.mark.asyncio
181 async def test_lookup_fallback(self):
182 task = _make_task()
183 task._session.lookup = AsyncMock(return_value="LOOKED_UP")
184 result = await task._resolve("site.i2p")
185 assert result == "LOOKED_UP"