distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for HTTPServerTask — TDD: tests before implementation."""
2
3import pytest
4
5from i2p_apps.i2ptunnel.config import TunnelDefinition, TunnelType
6from i2p_apps.i2ptunnel.http_server import HTTPServerTask
7
8
9def _make_server_def(**kw):
10 defaults = dict(
11 name="http-server",
12 type=TunnelType.HTTPSERVER,
13 target_host="127.0.0.1",
14 target_port=7658,
15 spoofed_host="mysite.i2p",
16 )
17 defaults.update(kw)
18 return TunnelDefinition(**defaults)
19
20
21class MockServerSession:
22 def __init__(self):
23 self.destination = "SERV" * 129
24 async def accept(self):
25 raise NotImplementedError
26 async def close(self):
27 pass
28
29
30class TestHostSpoofing:
31 def test_replace_host_header(self):
32 task = HTTPServerTask(_make_server_def(spoofed_host="mysite.i2p"), MockServerSession())
33 headers = {"Host": "original.example.com", "Content-Type": "text/html"}
34 spoofed = task._spoof_host(headers)
35 assert spoofed["Host"] == "mysite.i2p"
36 assert spoofed["Content-Type"] == "text/html"
37
38 def test_no_spoofed_host_preserves(self):
39 task = HTTPServerTask(_make_server_def(spoofed_host=""), MockServerSession())
40 headers = {"Host": "original.com"}
41 spoofed = task._spoof_host(headers)
42 assert spoofed["Host"] == "original.com"
43
44
45class TestI2PDestHeaders:
46 def test_inject_dest_headers(self):
47 task = HTTPServerTask(_make_server_def(), MockServerSession())
48 headers = {"Host": "mysite.i2p"}
49 remote_dest = "A" * 516
50 injected = task._inject_i2p_headers(headers, remote_dest)
51 assert "X-I2P-DestB64" in injected
52 assert injected["X-I2P-DestB64"] == remote_dest
53 assert "X-I2P-DestHash" in injected
54 assert "X-I2P-DestB32" in injected
55
56 def test_dest_hash_format(self):
57 task = HTTPServerTask(_make_server_def(), MockServerSession())
58 headers = {}
59 remote_dest = "A" * 516
60 injected = task._inject_i2p_headers(headers, remote_dest)
61 # Hash should be a hex string
62 assert len(injected["X-I2P-DestHash"]) == 64 # SHA256 hex
63
64
65class TestResponseStripping:
66 def test_strip_server_headers(self):
67 task = HTTPServerTask(_make_server_def(), MockServerSession())
68 headers = {
69 "Content-Type": "text/html",
70 "Server": "Apache/2.4",
71 "Date": "Mon, 01 Jan 2024 00:00:00 GMT",
72 "X-Powered-By": "PHP/8.0",
73 "X-Runtime": "0.123",
74 "Proxy": "evil",
75 }
76 stripped = task._strip_response_headers(headers)
77 assert "Content-Type" in stripped
78 assert "Server" not in stripped
79 assert "Date" not in stripped
80 assert "X-Powered-By" not in stripped
81 assert "X-Runtime" not in stripped
82 assert "Proxy" not in stripped
83
84 def test_preserves_content_length(self):
85 task = HTTPServerTask(_make_server_def(), MockServerSession())
86 headers = {"Content-Length": "1234", "Server": "nginx"}
87 stripped = task._strip_response_headers(headers)
88 assert stripped["Content-Length"] == "1234"
89
90
91class TestPostRateLimiting:
92 def test_post_allowed_under_limit(self):
93 task = HTTPServerTask(_make_server_def(), MockServerSession())
94 task._max_posts = 5
95 task._post_check_time = 60
96 assert task._check_post_limit("peer1") is True
97
98 def test_post_rejected_at_limit(self):
99 task = HTTPServerTask(_make_server_def(), MockServerSession())
100 task._max_posts = 2
101 task._post_check_time = 60
102 task._record_post("peer1")
103 task._record_post("peer1")
104 assert task._check_post_limit("peer1") is False
105
106 def test_post_global_limit(self):
107 task = HTTPServerTask(_make_server_def(), MockServerSession())
108 task._max_total_posts = 3
109 task._post_check_time = 60
110 task._record_post("peer1")
111 task._record_post("peer2")
112 task._record_post("peer3")
113 assert task._check_post_limit("peer4") is False
114
115
116class TestInproxyRejection:
117 def test_reject_via_header(self):
118 task = HTTPServerTask(_make_server_def(), MockServerSession())
119 task._reject_inproxy = True
120 headers = {"Via": "1.1 proxy.example.com", "Host": "mysite.i2p"}
121 assert task._is_inproxy_request(headers) is True
122
123 def test_reject_x_forwarded(self):
124 task = HTTPServerTask(_make_server_def(), MockServerSession())
125 task._reject_inproxy = True
126 headers = {"X-Forwarded-For": "1.2.3.4"}
127 assert task._is_inproxy_request(headers) is True
128
129 def test_allow_direct_request(self):
130 task = HTTPServerTask(_make_server_def(), MockServerSession())
131 task._reject_inproxy = True
132 headers = {"Host": "mysite.i2p"}
133 assert task._is_inproxy_request(headers) is False
134
135 def test_inproxy_check_disabled(self):
136 task = HTTPServerTask(_make_server_def(), MockServerSession())
137 task._reject_inproxy = False
138 headers = {"Via": "1.1 proxy"}
139 assert task._is_inproxy_request(headers) is False