Skip to content

The test-tunnel sample test tools

test_tunnel.cmd_test

A collection of sample command-line tools to test tunnel implementations.

Test the microsocks simple SOCKS5 server implementation

test_tunnel.cmd_test.microsocks

Run some tests on the microsocks proxy server and client.

ATYP_INET: Final = 1 module-attribute

The SOCKS5 address type for IPv4.

ATYP_INET6: Final = 4 module-attribute

The SOCKS5 address type for IPv6.

TestMicroSOCKS

Bases: TestTunnel

Run the tunnel tests using a microsocks server.

Source code in src/test_tunnel/cmd_test/microsocks.py 
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
class TestMicroSOCKS(run_test.TestTunnel):
    """Run the tunnel tests using a microsocks server."""

    def slug(self) -> str:
        """Identify the microsocks tunnel."""
        return "microsocks"

    @classmethod
    def quirk_server_returns_null_ipv4_response(cls) -> bool:
        """Expect 4 + 2 zeroes as a response to the "connect" request.

        The microsocks server does not bother returning any connection information in
        the response to the "connect" request; instead, it returns an address type of
        IPv4 and four + two bytes of zeroes.
        """
        return True

    def do_handshake(
        self,
        cli_sock: socket.socket,
        svc_listen: addresses.AddrPort,
    ) -> tuple[addresses.Address, int] | None:
        """Perform the SOCKS5 handshake."""

        def get_atyp(family: int) -> int:
            """Get the SOCKS5 "address type" value depending on the address family."""
            if family == socket.AF_INET:
                return ATYP_INET
            if family == socket.AF_INET6:
                return ATYP_INET6
            raise addresses.UnsupportedAddressFamilyError(family)

        def get_family(atyp: int) -> int:
            """Get the IP address family depending on the SOCKS5 "address type" value."""
            if atyp == ATYP_INET:
                return socket.AF_INET
            if atyp == ATYP_INET6:
                return socket.AF_INET6
            raise UnsupportedAddressTypeError(atyp)

        self.cfg.log.debug("Sending 'none' auth")
        cli_sock.send(bytes([5, 1, 0]))

        self.cfg.log.debug("Waiting for the server's auth response")
        self.expect_read(cli_sock, bytes([5, 0]), "auth response")

        family: Final = get_atyp(svc_listen.address.family)
        data: Final = (
            bytes([5, 1, 0, family])
            + svc_listen.address.packed
            + struct.pack(">h", svc_listen.svc_port)
        )
        self.cfg.log.debug("Sending a SOCKS5 CONNECT request: %(data)r", {"data": data})
        cli_sock.send(data)

        self.cfg.log.debug("Waiting for the server's connect response")
        ack = bytes([5, 0, 0])
        if self.quirk_server_returns_null_ipv4_response():
            ack += bytes([1] + [0] * 4 + [0] * 2)
            self.expect_read(cli_sock, ack, "connect response")
            return None

        ack += bytes([family])
        compare_len: Final = len(ack)
        ack += bytes([0] * len(svc_listen.address.packed) + [0] * 2)
        resp: Final = self.expect_read(cli_sock, ack, "connect response", compare_len=compare_len)
        res_family, buf_addr, buf_port = (
            get_family(resp[compare_len - 1]),
            resp[compare_len:-2],
            resp[-2:],
        )
        res_addr: Final = addresses.ipclass(res_family)(buf_addr)
        res_port: Final = struct.unpack(">h", buf_port)[0]
        return (
            addresses.Address(family=res_family, address=str(res_addr), packed=buf_addr),
            res_port,
        )

    @contextlib.contextmanager
    def do_spawn_server(
        self,
        proxy_listen: addresses.AddrPort,
        svc_listen: addresses.AddrPort,
    ) -> Iterator[subprocess.Popen[str]]:
        """Spawn the microsocks proxy process."""
        yield subprocess.Popen(  # noqa: S603
            [
                self.cfg.prog,
                "-i",
                proxy_listen.address.address,
                "-p",
                str(proxy_listen.svc_port),
                "-b",
                svc_listen.clients[0].address,
            ],
            bufsize=0,
            encoding="UTF-8",
            env=self.cfg.utf8_env,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )
do_handshake(cli_sock, svc_listen)

Perform the SOCKS5 handshake.

Source code in src/test_tunnel/cmd_test/microsocks.py 
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def do_handshake(
    self,
    cli_sock: socket.socket,
    svc_listen: addresses.AddrPort,
) -> tuple[addresses.Address, int] | None:
    """Perform the SOCKS5 handshake."""

    def get_atyp(family: int) -> int:
        """Get the SOCKS5 "address type" value depending on the address family."""
        if family == socket.AF_INET:
            return ATYP_INET
        if family == socket.AF_INET6:
            return ATYP_INET6
        raise addresses.UnsupportedAddressFamilyError(family)

    def get_family(atyp: int) -> int:
        """Get the IP address family depending on the SOCKS5 "address type" value."""
        if atyp == ATYP_INET:
            return socket.AF_INET
        if atyp == ATYP_INET6:
            return socket.AF_INET6
        raise UnsupportedAddressTypeError(atyp)

    self.cfg.log.debug("Sending 'none' auth")
    cli_sock.send(bytes([5, 1, 0]))

    self.cfg.log.debug("Waiting for the server's auth response")
    self.expect_read(cli_sock, bytes([5, 0]), "auth response")

    family: Final = get_atyp(svc_listen.address.family)
    data: Final = (
        bytes([5, 1, 0, family])
        + svc_listen.address.packed
        + struct.pack(">h", svc_listen.svc_port)
    )
    self.cfg.log.debug("Sending a SOCKS5 CONNECT request: %(data)r", {"data": data})
    cli_sock.send(data)

    self.cfg.log.debug("Waiting for the server's connect response")
    ack = bytes([5, 0, 0])
    if self.quirk_server_returns_null_ipv4_response():
        ack += bytes([1] + [0] * 4 + [0] * 2)
        self.expect_read(cli_sock, ack, "connect response")
        return None

    ack += bytes([family])
    compare_len: Final = len(ack)
    ack += bytes([0] * len(svc_listen.address.packed) + [0] * 2)
    resp: Final = self.expect_read(cli_sock, ack, "connect response", compare_len=compare_len)
    res_family, buf_addr, buf_port = (
        get_family(resp[compare_len - 1]),
        resp[compare_len:-2],
        resp[-2:],
    )
    res_addr: Final = addresses.ipclass(res_family)(buf_addr)
    res_port: Final = struct.unpack(">h", buf_port)[0]
    return (
        addresses.Address(family=res_family, address=str(res_addr), packed=buf_addr),
        res_port,
    )
do_spawn_server(proxy_listen, svc_listen)

Spawn the microsocks proxy process.

Source code in src/test_tunnel/cmd_test/microsocks.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
@contextlib.contextmanager
def do_spawn_server(
    self,
    proxy_listen: addresses.AddrPort,
    svc_listen: addresses.AddrPort,
) -> Iterator[subprocess.Popen[str]]:
    """Spawn the microsocks proxy process."""
    yield subprocess.Popen(  # noqa: S603
        [
            self.cfg.prog,
            "-i",
            proxy_listen.address.address,
            "-p",
            str(proxy_listen.svc_port),
            "-b",
            svc_listen.clients[0].address,
        ],
        bufsize=0,
        encoding="UTF-8",
        env=self.cfg.utf8_env,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )
quirk_server_returns_null_ipv4_response() classmethod

Expect 4 + 2 zeroes as a response to the "connect" request.

The microsocks server does not bother returning any connection information in the response to the "connect" request; instead, it returns an address type of IPv4 and four + two bytes of zeroes.

Source code in src/test_tunnel/cmd_test/microsocks.py 
51
52
53
54
55
56
57
58
59
@classmethod
def quirk_server_returns_null_ipv4_response(cls) -> bool:
    """Expect 4 + 2 zeroes as a response to the "connect" request.

    The microsocks server does not bother returning any connection information in
    the response to the "connect" request; instead, it returns an address type of
    IPv4 and four + two bytes of zeroes.
    """
    return True
slug()

Identify the microsocks tunnel.

Source code in src/test_tunnel/cmd_test/microsocks.py 
47
48
49
def slug(self) -> str:
    """Identify the microsocks tunnel."""
    return "microsocks"

UnsupportedAddressTypeError dataclass

Bases: Exception

An unsupported address family was specified.

Source code in src/test_tunnel/cmd_test/microsocks.py 
33
34
35
36
37
38
39
40
41
@dataclasses.dataclass
class UnsupportedAddressTypeError(Exception):
    """An unsupported address family was specified."""

    atyp: int

    def __str__(self) -> str:
        """Provide a human-readable description of the error."""
        return f"Unsupported SOCKS5 address type {self.atyp}"
__str__()

Provide a human-readable description of the error.

Source code in src/test_tunnel/cmd_test/microsocks.py 
39
40
41
def __str__(self) -> str:
    """Provide a human-readable description of the error."""
    return f"Unsupported SOCKS5 address type {self.atyp}"

main(cfg)

Parse command-line arguments, prepare the environment, run tests.

Source code in src/test_tunnel/cmd_test/microsocks.py 
147
148
149
150
151
@cmd_util.click_common_args("microsocks")
def main(cfg: defs.ConfigProg) -> None:
    """Parse command-line arguments, prepare the environment, run tests."""
    tester: Final = TestMicroSOCKS(cfg)
    tester.run()

Test the socat tool's TCP connection forwarding

test_tunnel.cmd_test.socat

Run some tests using the socat tool in forwarding mode.

TestSoCat

Bases: TestTunnel

Run the tunnel tests using a socat instance.

Source code in src/test_tunnel/cmd_test/socat.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
class TestSoCat(run_test.TestTunnel):
    """Run the tunnel tests using a socat instance."""

    def slug(self) -> str:
        """Identify the socat tool."""
        return "socat"

    def do_handshake(
        self,
        cli_sock: socket.socket,
        svc_listen: addresses.AddrPort,
    ) -> tuple[addresses.Address, int] | None:
        """No handshake for socat."""
        self.cfg.log.info("No handshake necessary for socat")
        self.cfg.log.debug(
            "Nothing to do for a %(cli)s / %(svc)s connection",
            {"cli": cli_sock, "svc": svc_listen},
        )
        return None

    @contextlib.contextmanager
    def do_spawn_server(
        self,
        proxy_listen: addresses.AddrPort,
        svc_listen: addresses.AddrPort,
    ) -> Iterator[subprocess.Popen[str]]:
        """Spawn the socat proxy process."""
        yield subprocess.Popen(  # noqa: S603
            [
                self.cfg.prog,
                "-v",
                (
                    f"TCP{addresses.family_id(proxy_listen.address.family)}-LISTEN:"
                    f"{proxy_listen.svc_port},"
                    f"bind=[{proxy_listen.address.address}],reuseaddr,fork"
                ),
                (
                    f"TCP{addresses.family_id(svc_listen.clients[0].family)}:"
                    f"[{svc_listen.address.address}]:{svc_listen.svc_port},"
                    f"bind=[{svc_listen.clients[0].address}]"
                ),
            ],
            bufsize=0,
            encoding="UTF-8",
            env=self.cfg.utf8_env,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )
do_handshake(cli_sock, svc_listen)

No handshake for socat.

Source code in src/test_tunnel/cmd_test/socat.py 
31
32
33
34
35
36
37
38
39
40
41
42
def do_handshake(
    self,
    cli_sock: socket.socket,
    svc_listen: addresses.AddrPort,
) -> tuple[addresses.Address, int] | None:
    """No handshake for socat."""
    self.cfg.log.info("No handshake necessary for socat")
    self.cfg.log.debug(
        "Nothing to do for a %(cli)s / %(svc)s connection",
        {"cli": cli_sock, "svc": svc_listen},
    )
    return None
do_spawn_server(proxy_listen, svc_listen)

Spawn the socat proxy process.

Source code in src/test_tunnel/cmd_test/socat.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
@contextlib.contextmanager
def do_spawn_server(
    self,
    proxy_listen: addresses.AddrPort,
    svc_listen: addresses.AddrPort,
) -> Iterator[subprocess.Popen[str]]:
    """Spawn the socat proxy process."""
    yield subprocess.Popen(  # noqa: S603
        [
            self.cfg.prog,
            "-v",
            (
                f"TCP{addresses.family_id(proxy_listen.address.family)}-LISTEN:"
                f"{proxy_listen.svc_port},"
                f"bind=[{proxy_listen.address.address}],reuseaddr,fork"
            ),
            (
                f"TCP{addresses.family_id(svc_listen.clients[0].family)}:"
                f"[{svc_listen.address.address}]:{svc_listen.svc_port},"
                f"bind=[{svc_listen.clients[0].address}]"
            ),
        ],
        bufsize=0,
        encoding="UTF-8",
        env=self.cfg.utf8_env,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )
slug()

Identify the socat tool.

Source code in src/test_tunnel/cmd_test/socat.py 
27
28
29
def slug(self) -> str:
    """Identify the socat tool."""
    return "socat"

main(cfg)

Parse command-line arguments, prepare the environment, run tests.

Source code in src/test_tunnel/cmd_test/socat.py 
74
75
76
77
78
@cmd_util.click_common_args("socat")
def main(cfg: defs.ConfigProg) -> None:
    """Parse command-line arguments, prepare the environment, run tests."""
    tester: Final = TestSoCat(cfg)
    tester.run()