distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for SOCKSProxy server — _handle_client and start/stop."""
2
3import asyncio
4from unittest.mock import AsyncMock, MagicMock
5
6import pytest
7
8from i2p_apps.i2ptunnel.socks_proxy import (
9 SOCKSProxy, parse_socks_greeting, parse_socks_request,
10 build_socks_greeting_reply, build_socks_reply,
11 SOCKS_VERSION, AUTH_NONE, REPLY_HOST_UNREACHABLE,
12)
13
14
15def _mock_writer():
16 writer = MagicMock()
17 writer.write = MagicMock()
18 writer.drain = AsyncMock()
19 writer.close = MagicMock()
20 writer.wait_closed = AsyncMock()
21 return writer
22
23
24class TestSOCKSProxyProperties:
25 def test_defaults(self):
26 proxy = SOCKSProxy()
27 assert proxy.listen_address == ("127.0.0.1", 4445)
28 assert proxy.is_running is False
29
30 def test_custom(self):
31 proxy = SOCKSProxy("0.0.0.0", 9050)
32 assert proxy.listen_address == ("0.0.0.0", 9050)
33
34 @pytest.mark.asyncio
35 async def test_start_stop(self):
36 proxy = SOCKSProxy("127.0.0.1", 0)
37 await proxy.start()
38 assert proxy.is_running is True
39 await proxy.stop()
40 assert proxy.is_running is False
41
42 @pytest.mark.asyncio
43 async def test_stop_without_start(self):
44 proxy = SOCKSProxy()
45 await proxy.stop()
46 assert proxy.is_running is False
47
48
49class TestSOCKSProxyHandleClient:
50 @pytest.mark.asyncio
51 async def test_valid_connect(self):
52 """Full SOCKS5 handshake: greeting → reply → request → host unreachable."""
53 proxy = SOCKSProxy()
54
55 # Build greeting: VER=5, NMETHODS=1, METHODS=[0x00]
56 greeting = bytes([SOCKS_VERSION, 1, AUTH_NONE])
57 # Build request: VER=5, CMD=1 (CONNECT), RSV=0, ATYP=3 (domain)
58 domain = b"example.i2p"
59 request = bytes([SOCKS_VERSION, 1, 0, 3, len(domain)]) + domain + b"\x01\xbb" # port 443
60
61 read_count = 0
62 async def mock_read(n):
63 nonlocal read_count
64 read_count += 1
65 if read_count == 1:
66 return greeting
67 return request
68
69 reader = AsyncMock()
70 reader.read = mock_read
71 writer = _mock_writer()
72
73 await proxy._handle_client(reader, writer)
74
75 # Should have sent greeting reply then SOCKS reply
76 calls = writer.write.call_args_list
77 assert len(calls) == 2
78 # First call: greeting reply
79 assert calls[0][0][0] == build_socks_greeting_reply(AUTH_NONE)
80 # Second call: host unreachable reply
81 assert calls[1][0][0] == build_socks_reply(REPLY_HOST_UNREACHABLE)
82
83 @pytest.mark.asyncio
84 async def test_no_acceptable_auth(self):
85 """Client offers only auth methods we don't support → 0xFF reply."""
86 proxy = SOCKSProxy()
87
88 greeting = bytes([SOCKS_VERSION, 1, 0x02]) # only USERNAME/PASSWORD
89
90 reader = AsyncMock()
91 reader.read = AsyncMock(return_value=greeting)
92 writer = _mock_writer()
93
94 await proxy._handle_client(reader, writer)
95
96 calls = writer.write.call_args_list
97 assert len(calls) == 1
98 assert calls[0][0][0] == build_socks_greeting_reply(0xFF)
99 writer.close.assert_called()
100
101 @pytest.mark.asyncio
102 async def test_exception_closes_writer(self):
103 proxy = SOCKSProxy()
104
105 reader = AsyncMock()
106 reader.read = AsyncMock(side_effect=ConnectionResetError("reset"))
107 writer = _mock_writer()
108
109 await proxy._handle_client(reader, writer)
110 writer.close.assert_called()
111
112 @pytest.mark.asyncio
113 async def test_invalid_greeting(self):
114 """Too-short greeting → ValueError → exception path."""
115 proxy = SOCKSProxy()
116
117 reader = AsyncMock()
118 reader.read = AsyncMock(return_value=b"\x04") # SOCKS4, not 5
119 writer = _mock_writer()
120
121 await proxy._handle_client(reader, writer)
122 writer.close.assert_called()