"""Tests for I2P tunnel infrastructure — HTTP proxy, SOCKS proxy, server tunnels.""" import struct import pytest from i2p_apps.i2ptunnel.http_proxy import ( HTTPRequest, HTTPProxy, parse_http_request, extract_i2p_destination, ) from i2p_apps.i2ptunnel.socks_proxy import ( SOCKS_VERSION, AUTH_NONE, CMD_CONNECT, ATYP_DOMAIN, REPLY_SUCCESS, REPLY_GENERAL_FAILURE, REPLY_HOST_UNREACHABLE, SOCKSGreeting, SOCKSRequest, SOCKSProxy, parse_socks_greeting, build_socks_greeting_reply, parse_socks_request, build_socks_reply, ) from i2p_apps.i2ptunnel.server_tunnel import ServerTunnel from i2p_apps.i2ptunnel.tunnel_manager import I2PTunnelManager # --------------------------------------------------------------------------- # HTTPProxy tests # --------------------------------------------------------------------------- class TestParseHTTPRequest: """Test HTTP request first-line parsing.""" def test_connect_request(self): req = parse_http_request("CONNECT example.i2p:443 HTTP/1.1") assert req.method == "CONNECT" assert req.host == "example.i2p" assert req.port == 443 assert req.is_connect is True assert req.raw_first_line == "CONNECT example.i2p:443 HTTP/1.1" def test_regular_get_request(self): req = parse_http_request("GET http://forum.i2p/index.html HTTP/1.1") assert req.method == "GET" assert req.host == "forum.i2p" assert req.port == 80 assert req.path == "/index.html" assert req.is_connect is False def test_regular_get_with_port(self): req = parse_http_request("GET http://forum.i2p:8080/api HTTP/1.1") assert req.host == "forum.i2p" assert req.port == 8080 assert req.path == "/api" def test_connect_without_port(self): req = parse_http_request("CONNECT example.i2p HTTP/1.1") assert req.host == "example.i2p" assert req.port == 443 # default for CONNECT def test_post_request(self): req = parse_http_request("POST http://mail.i2p/send HTTP/1.1") assert req.method == "POST" assert req.host == "mail.i2p" assert req.path == "/send" assert req.is_connect is False class TestExtractI2PDestination: """Test .i2p destination extraction.""" def test_plain_i2p_host(self): assert extract_i2p_destination("forum.i2p") == "forum.i2p" def test_subdomain_i2p(self): assert extract_i2p_destination("www.forum.i2p") == "www.forum.i2p" def test_non_i2p_host(self): assert extract_i2p_destination("example.com") is None def test_empty_host(self): assert extract_i2p_destination("") is None def test_i2p_in_middle(self): # "i2p.example.com" is NOT an i2p destination assert extract_i2p_destination("i2p.example.com") is None class TestHTTPProxy: """Test HTTPProxy instance behavior.""" def test_default_listen_address(self): proxy = HTTPProxy() assert proxy.listen_address == ("127.0.0.1", 4444) def test_custom_listen_address(self): proxy = HTTPProxy(listen_host="0.0.0.0", listen_port=8888) assert proxy.listen_address == ("0.0.0.0", 8888) def test_not_running_initially(self): proxy = HTTPProxy() assert proxy.is_running is False def test_build_502_response(self): proxy = HTTPProxy() resp = proxy.build_error_response(502, "Bad Gateway") assert b"502" in resp assert b"Bad Gateway" in resp assert resp.startswith(b"HTTP/1.1 502") # --------------------------------------------------------------------------- # SOCKSProxy tests # --------------------------------------------------------------------------- class TestParseSOCKSGreeting: """Test SOCKS5 greeting parsing.""" def test_no_auth_greeting(self): # version=5, nmethods=1, methods=[0 (no auth)] data = bytes([0x05, 0x01, 0x00]) greeting = parse_socks_greeting(data) assert greeting.version == 5 assert greeting.methods == [AUTH_NONE] def test_multi_method_greeting(self): # version=5, nmethods=2, methods=[0, 2] data = bytes([0x05, 0x02, 0x00, 0x02]) greeting = parse_socks_greeting(data) assert greeting.version == 5 assert greeting.methods == [0, 2] def test_invalid_version_raises(self): data = bytes([0x04, 0x01, 0x00]) # SOCKS4 with pytest.raises(ValueError, match="[Vv]ersion"): parse_socks_greeting(data) class TestBuildSOCKSGreetingReply: """Test SOCKS5 greeting reply construction.""" def test_no_auth_reply(self): reply = build_socks_greeting_reply(AUTH_NONE) assert reply == bytes([0x05, 0x00]) def test_custom_method_reply(self): reply = build_socks_greeting_reply(0x02) assert reply == bytes([0x05, 0x02]) class TestParseSOCKSRequest: """Test SOCKS5 CONNECT request parsing.""" def test_domain_connect(self): # VER=5, CMD=1 (connect), RSV=0, ATYP=3 (domain) domain = b"forum.i2p" data = bytes([0x05, 0x01, 0x00, 0x03, len(domain)]) + domain + struct.pack("!H", 80) req = parse_socks_request(data) assert req.version == 5 assert req.command == CMD_CONNECT assert req.address_type == ATYP_DOMAIN assert req.dest_addr == "forum.i2p" assert req.dest_port == 80 def test_domain_with_high_port(self): domain = b"mail.i2p" data = bytes([0x05, 0x01, 0x00, 0x03, len(domain)]) + domain + struct.pack("!H", 8443) req = parse_socks_request(data) assert req.dest_addr == "mail.i2p" assert req.dest_port == 8443 def test_invalid_version_raises(self): data = bytes([0x04, 0x01, 0x00, 0x03, 0x05]) + b"a.i2p" + struct.pack("!H", 80) with pytest.raises(ValueError, match="[Vv]ersion"): parse_socks_request(data) class TestBuildSOCKSReply: """Test SOCKS5 reply construction.""" def test_success_reply(self): reply = build_socks_reply(REPLY_SUCCESS) assert reply[0] == 0x05 # version assert reply[1] == REPLY_SUCCESS assert reply[2] == 0x00 # reserved assert len(reply) == 10 # VER + REP + RSV + ATYP(1) + ADDR(4) + PORT(2) def test_failure_reply(self): reply = build_socks_reply(REPLY_GENERAL_FAILURE) assert reply[1] == REPLY_GENERAL_FAILURE def test_host_unreachable_reply(self): reply = build_socks_reply(REPLY_HOST_UNREACHABLE) assert reply[1] == REPLY_HOST_UNREACHABLE def test_custom_bind(self): reply = build_socks_reply(REPLY_SUCCESS, bind_addr="127.0.0.1", bind_port=1080) port_bytes = reply[-2:] port = struct.unpack("!H", port_bytes)[0] assert port == 1080 class TestSOCKSProxy: """Test SOCKSProxy instance.""" def test_default_listen_address(self): proxy = SOCKSProxy() assert proxy.listen_address == ("127.0.0.1", 4445) def test_not_running_initially(self): proxy = SOCKSProxy() assert proxy.is_running is False # --------------------------------------------------------------------------- # ServerTunnel tests # --------------------------------------------------------------------------- class TestServerTunnel: """Test server tunnel dataclass.""" def test_create_with_local_address(self): tunnel = ServerTunnel(name="web", local_host="127.0.0.1", local_port=8080) assert tunnel.name == "web" assert tunnel.local_host == "127.0.0.1" assert tunnel.local_port == 8080 assert tunnel.i2p_port == 0 assert tunnel.running is False def test_create_with_custom_i2p_port(self): tunnel = ServerTunnel(name="irc", local_host="127.0.0.1", local_port=6667, i2p_port=6668) assert tunnel.i2p_port == 6668 def test_forward_data_mock(self): """ServerTunnel should track forwarded byte counts.""" tunnel = ServerTunnel(name="web", local_host="127.0.0.1", local_port=8080) assert tunnel.bytes_forwarded == 0 tunnel.record_forwarded(1024) assert tunnel.bytes_forwarded == 1024 tunnel.record_forwarded(512) assert tunnel.bytes_forwarded == 1536 # --------------------------------------------------------------------------- # I2PTunnelManager tests # --------------------------------------------------------------------------- class TestI2PTunnelManager: """Test tunnel manager add/remove/list.""" def test_add_and_list(self): mgr = I2PTunnelManager() proxy = HTTPProxy() mgr.add_tunnel("http", proxy) assert mgr.list_tunnels() == ["http"] assert mgr.tunnel_count == 1 def test_remove_tunnel(self): mgr = I2PTunnelManager() proxy = HTTPProxy() mgr.add_tunnel("http", proxy) assert mgr.remove_tunnel("http") is True assert mgr.tunnel_count == 0 def test_remove_nonexistent(self): mgr = I2PTunnelManager() assert mgr.remove_tunnel("nope") is False def test_get_tunnel_by_name(self): mgr = I2PTunnelManager() proxy = SOCKSProxy() mgr.add_tunnel("socks", proxy) assert mgr.get_tunnel("socks") is proxy def test_get_tunnel_missing(self): mgr = I2PTunnelManager() assert mgr.get_tunnel("missing") is None def test_add_duplicate_raises(self): mgr = I2PTunnelManager() mgr.add_tunnel("http", HTTPProxy()) with pytest.raises(ValueError, match="already exists"): mgr.add_tunnel("http", HTTPProxy()) def test_mixed_tunnel_types(self): mgr = I2PTunnelManager() mgr.add_tunnel("http", HTTPProxy()) mgr.add_tunnel("socks", SOCKSProxy()) mgr.add_tunnel("server", ServerTunnel(name="web", local_host="127.0.0.1", local_port=80)) assert mgr.tunnel_count == 3 assert sorted(mgr.list_tunnels()) == ["http", "server", "socks"]