pyssh-client/tests/test_socks_tunnel.py

"""Tunnel dynamique (-D) : verifie que le proxy SOCKS5 / SOCKS4a negocie
correctement la destination demandee par le client puis relaie le trafic a
travers un (faux) transport SSH -- sans connexion SSH reelle, comme le fait
deja test_tunnels.py pour la redirection locale."""
import pathlib
import sys

sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent))

import socket
import threading
import time

from app.tunnels import DIRECTION_DYNAMIC, TunnelManager


class EchoServer(threading.Thread):
    """Serveur de destination : renvoie ce qu'il recoit, prefixe."""

    def __init__(self):
        super().__init__(daemon=True)
        self.sock = socket.socket()
        self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.sock.bind(("127.0.0.1", 0))
        self.sock.listen(5)
        self.port = self.sock.getsockname()[1]

    def run(self):
        while True:
            try:
                conn, _ = self.sock.accept()
            except OSError:
                return
            data = conn.recv(1024)
            conn.sendall(b"echo:" + data)
            conn.close()


class FakeTransport:
    """Imite paramiko.Transport : ouvre un vrai socket vers la destination negociee."""

    def open_channel(self, kind, dest, src):
        assert kind == "direct-tcpip"
        return socket.create_connection(dest, timeout=5)


def socks5_connect(proxy_port: int, host: str, port: int) -> socket.socket:
    sock = socket.create_connection(("127.0.0.1", proxy_port), timeout=5)
    sock.sendall(b"\x05\x01\x00")  # version 5, 1 methode proposee : "sans authentification"
    assert sock.recv(2) == b"\x05\x00", "methode 'sans authentification' attendue"
    sock.sendall(b"\x05\x01\x00\x01" + socket.inet_aton(host) + port.to_bytes(2, "big"))
    reply = sock.recv(10)
    assert reply[0:2] == b"\x05\x00", f"CONNECT refuse : {reply}"
    return sock


def socks4a_connect(proxy_port: int, host: str, port: int) -> socket.socket:
    sock = socket.create_connection(("127.0.0.1", proxy_port), timeout=5)
    # IP 0.0.0.1 = extension SOCKS4a : le nom suit l'identifiant utilisateur
    sock.sendall(
        b"\x04\x01" + port.to_bytes(2, "big") + b"\x00\x00\x00\x01"
        + b"user\x00" + host.encode("ascii") + b"\x00"
    )
    reply = sock.recv(8)
    assert reply[0:2] == b"\x00\x5a", f"CONNECT refuse : {reply}"
    return sock


echo = EchoServer()
echo.start()
print(f"serveur de destination (echo) sur le port {echo.port}")

manager = TunnelManager(FakeTransport())
tunnel = manager.start({"direction": DIRECTION_DYNAMIC, "listen_port": 0})
assert tunnel.error is None, tunnel.error
proxy_port = tunnel._server.server_address[1]
print("proxy SOCKS ouvert sur le port", proxy_port)
time.sleep(0.2)

print()
print("=== 1. SOCKS5, adresse IPv4 ===")
client = socks5_connect(proxy_port, "127.0.0.1", echo.port)
client.sendall(b"bonjour-socks5")
reponse = client.recv(1024)
client.close()
print("  recu:", reponse)
assert reponse == b"echo:bonjour-socks5", reponse

print()
print("=== 2. SOCKS4a, nom de domaine ===")
client = socks4a_connect(proxy_port, "localhost", echo.port)
client.sendall(b"bonjour-socks4a")
reponse = client.recv(1024)
client.close()
print("  recu:", reponse)
assert reponse == b"echo:bonjour-socks4a", reponse

print()
print("=== 3. SOCKS5 : commande non supportee (BIND) rejetee proprement ===")
sock = socket.create_connection(("127.0.0.1", proxy_port), timeout=5)
sock.sendall(b"\x05\x01\x00")
assert sock.recv(2) == b"\x05\x00"
sock.sendall(b"\x05\x02\x00\x01" + socket.inet_aton("127.0.0.1") + (1234).to_bytes(2, "big"))  # BIND
reply = sock.recv(10)
print("  reponse:", reply)
assert reply[1] == 0x07, reply  # 0x07 = commande non supportee
sock.close()
print("  OK")

print()
print("=== 4. Fermeture du tunnel : le port SOCKS n'accepte plus de connexion ===")
manager.stop_all()
time.sleep(0.3)
try:
    socket.create_connection(("127.0.0.1", proxy_port), timeout=1).close()
    ferme = False
except OSError:
    ferme = True
assert ferme, "le port doit etre libere apres fermeture"
print("  OK")

print()
print("SOCKS DYNAMIQUE (-D) : TOUS LES TESTS OK")