pyssh-client/app/tunnels.py

"""Tunnels SSH (redirection de ports), facon `ssh -L`, `ssh -R` et `ssh -D`.

Redirection locale (-L) : un port ouvert sur votre poste est relaye, a travers la
connexion SSH, vers un hote joignable depuis le serveur. Typiquement pour atteindre
l'interface web d'un service qui n'ecoute que sur la boucle locale du serveur.

Redirection distante (-R) : l'inverse, un port ouvert sur le serveur est relaye vers
une machine joignable depuis votre poste.

Redirection dynamique (-D) : un port ouvert sur votre poste se comporte comme un
proxy SOCKS ; la destination n'est pas fixee a l'avance mais negociee par chaque
client SOCKS (navigateur, curl...), puis atteinte a travers la connexion SSH.

Chaque tunnel tourne dans son propre thread : un relais bloque ne doit gener ni
l'interface ni les autres tunnels.
"""
from __future__ import annotations

import select
import socket
import socketserver
import threading

import paramiko

DIRECTION_LOCAL = "local"
DIRECTION_REMOTE = "remote"
DIRECTION_DYNAMIC = "dynamic"

DIRECTION_LABELS = {
    DIRECTION_LOCAL: "Locale  (-L)",
    DIRECTION_REMOTE: "Distante (-R)",
    DIRECTION_DYNAMIC: "Dynamique - SOCKS (-D)",
}


def describe(tunnel: dict) -> str:
    """Resume lisible, dans le sens de lecture du flux."""
    if tunnel.get("direction") == DIRECTION_REMOTE:
        return (f"serveur:{tunnel['listen_port']}  ->  "
                f"{tunnel['dest_host']}:{tunnel['dest_port']} (via ce poste)")
    if tunnel.get("direction") == DIRECTION_DYNAMIC:
        return f"SOCKS sur 127.0.0.1:{tunnel['listen_port']}  (destination negociee par le client)"
    return (f"localhost:{tunnel['listen_port']}  ->  "
            f"{tunnel['dest_host']}:{tunnel['dest_port']} (via le serveur)")


def _relay(sock: socket.socket, chan: paramiko.Channel) -> None:
    """Recopie les octets dans les deux sens jusqu'a fermeture."""
    try:
        while True:
            readable, _, _ = select.select([sock, chan], [], [], 1.0)
            if sock in readable:
                data = sock.recv(4096)
                if not data:
                    break
                chan.sendall(data)
            if chan in readable:
                data = chan.recv(4096)
                if not data:
                    break
                sock.sendall(data)
    except OSError:
        pass
    finally:
        for closeable in (chan, sock):
            try:
                closeable.close()
            except Exception:
                pass


class _ForwardServer(socketserver.ThreadingTCPServer):
    daemon_threads = True
    allow_reuse_address = True


class _ForwardHandler(socketserver.BaseRequestHandler):
    transport: paramiko.Transport
    dest_host: str
    dest_port: int

    def handle(self):
        try:
            chan = self.transport.open_channel(
                "direct-tcpip", (self.dest_host, self.dest_port), self.request.getpeername()
            )
        except Exception:
            return
        if chan is None:
            return
        _relay(self.request, chan)


# ----------------------------------------------------------------------------
# Redirection dynamique (-D) : proxy SOCKS local
# ----------------------------------------------------------------------------
def _recv_exact(sock: socket.socket, n: int) -> bytes:
    """recv() ne garantit pas de renvoyer les n octets demandes en un seul appel."""
    data = b""
    while len(data) < n:
        chunk = sock.recv(n - len(data))
        if not chunk:
            raise ConnectionError("Connexion fermee pendant la negociation SOCKS")
        data += chunk
    return data


def _read_c_string(sock: socket.socket) -> bytes:
    out = bytearray()
    while True:
        b = _recv_exact(sock, 1)
        if b == b"\x00":
            return bytes(out)
        out += b


def _socks5_handshake(sock: socket.socket) -> tuple[str, int] | None:
    nmethods = _recv_exact(sock, 1)[0]
    methods = _recv_exact(sock, nmethods) if nmethods else b""
    if 0x00 not in methods:
        sock.sendall(b"\x05\xff")  # aucune methode acceptable : on ne gere que "sans authentification"
        return None
    sock.sendall(b"\x05\x00")

    _, cmd, _, atyp = _recv_exact(sock, 4)
    if cmd != 0x01:  # seul CONNECT est pris en charge (pas BIND / UDP ASSOCIATE)
        sock.sendall(b"\x05\x07\x00\x01" + b"\x00" * 6)
        return None

    if atyp == 0x01:
        host = socket.inet_ntoa(_recv_exact(sock, 4))
    elif atyp == 0x03:
        length = _recv_exact(sock, 1)[0]
        host = _recv_exact(sock, length).decode("ascii", errors="replace")
    elif atyp == 0x04:
        host = socket.inet_ntop(socket.AF_INET6, _recv_exact(sock, 16))
    else:
        sock.sendall(b"\x05\x08\x00\x01" + b"\x00" * 6)  # type d'adresse non supporte
        return None
    port = int.from_bytes(_recv_exact(sock, 2), "big")

    # Adresse/port de retour a 0 : sans interet pour un CONNECT, la plupart des
    # clients ne les utilisent pas au-dela de verifier le code de succes.
    sock.sendall(b"\x05\x00\x00\x01" + b"\x00\x00\x00\x00\x00\x00")
    return host, port


def _socks4_handshake(sock: socket.socket) -> tuple[str, int] | None:
    cmd = _recv_exact(sock, 1)[0]
    port = int.from_bytes(_recv_exact(sock, 2), "big")
    ip_bytes = _recv_exact(sock, 4)
    _read_c_string(sock)  # identifiant utilisateur, ignore
    if cmd != 0x01:
        sock.sendall(b"\x00\x5b" + b"\x00" * 6)
        return None
    # SOCKS4a : une IP de la forme 0.0.0.x (x != 0) signale un nom de domaine a
    # lire ensuite plutot qu'une adresse IP directement utilisable.
    if ip_bytes[:3] == b"\x00\x00\x00" and ip_bytes[3] != 0:
        host = _read_c_string(sock).decode("ascii", errors="replace")
    else:
        host = socket.inet_ntoa(ip_bytes)
    sock.sendall(b"\x00\x5a" + port.to_bytes(2, "big") + ip_bytes)
    return host, port


def _socks_handshake(sock: socket.socket) -> tuple[str, int] | None:
    """Negocie une requete SOCKS4/4a ou SOCKS5 (CONNECT uniquement, sans
    authentification) et renvoie (hote, port) demande par le client, ou None si
    la negociation echoue ou porte sur une commande non prise en charge."""
    version = _recv_exact(sock, 1)
    if version == b"\x05":
        return _socks5_handshake(sock)
    if version == b"\x04":
        return _socks4_handshake(sock)
    return None


class _SocksHandler(socketserver.BaseRequestHandler):
    transport: paramiko.Transport

    def handle(self):
        sock = self.request
        try:
            dest = _socks_handshake(sock)
        except (OSError, ConnectionError, IndexError):
            return
        if dest is None:
            return
        try:
            chan = self.transport.open_channel("direct-tcpip", dest, sock.getpeername())
        except Exception:
            return
        if chan is None:
            return
        _relay(sock, chan)


class Tunnel:
    """Un tunnel actif, avec de quoi l'arreter."""

    def __init__(self, config: dict):
        self.config = config
        self.error: str | None = None
        self._server: _ForwardServer | None = None
        self._thread: threading.Thread | None = None
        self._transport: paramiko.Transport | None = None
        self._stop = threading.Event()

    @property
    def direction(self) -> str:
        return self.config.get("direction", DIRECTION_LOCAL)

    def start(self, transport: paramiko.Transport) -> None:
        self._transport = transport
        if self.direction == DIRECTION_REMOTE:
            self._start_remote(transport)
        elif self.direction == DIRECTION_DYNAMIC:
            self._start_dynamic(transport)
        else:
            self._start_local(transport)

    # ---------- redirection dynamique (-D) : proxy SOCKS ----------
    def _start_dynamic(self, transport: paramiko.Transport) -> None:
        config = self.config
        handler = type("SocksHandler", (_SocksHandler,), {"transport": transport})
        bind = config.get("listen_host") or "127.0.0.1"
        self._server = _ForwardServer((bind, int(config["listen_port"])), handler)
        self._thread = threading.Thread(target=self._server.serve_forever, daemon=True)
        self._thread.start()

    # ---------- redirection locale (-L) ----------
    def _start_local(self, transport: paramiko.Transport) -> None:
        config = self.config
        handler = type(
            "Handler", (_ForwardHandler,),
            {"transport": transport,
             "dest_host": config["dest_host"],
             "dest_port": int(config["dest_port"])},
        )
        bind = config.get("listen_host") or "127.0.0.1"
        self._server = _ForwardServer((bind, int(config["listen_port"])), handler)
        self._thread = threading.Thread(target=self._server.serve_forever, daemon=True)
        self._thread.start()

    # ---------- redirection distante (-R) ----------
    def _start_remote(self, transport: paramiko.Transport) -> None:
        config = self.config
        bind = config.get("listen_host") or ""
        transport.request_port_forward(bind, int(config["listen_port"]))
        self._thread = threading.Thread(target=self._accept_remote, args=(transport,), daemon=True)
        self._thread.start()

    def _accept_remote(self, transport: paramiko.Transport) -> None:
        while not self._stop.is_set():
            chan = transport.accept(1)
            if chan is None:
                continue
            try:
                sock = socket.create_connection(
                    (self.config["dest_host"], int(self.config["dest_port"])), timeout=10
                )
            except OSError:
                try:
                    chan.close()
                except Exception:
                    pass
                continue
            threading.Thread(target=_relay, args=(sock, chan), daemon=True).start()

    def stop(self) -> None:
        self._stop.set()
        if self._server:
            try:
                self._server.shutdown()
                self._server.server_close()
            except Exception:
                pass
            self._server = None
        if self.direction == DIRECTION_REMOTE and self._transport:
            try:
                self._transport.cancel_port_forward(
                    self.config.get("listen_host") or "", int(self.config["listen_port"])
                )
            except Exception:
                pass


class TunnelManager:
    """Tunnels d'une session SSH donnee."""

    def __init__(self, transport: paramiko.Transport):
        self.transport = transport
        self.tunnels: list[Tunnel] = []

    def start(self, config: dict) -> Tunnel:
        tunnel = Tunnel(config)
        try:
            tunnel.start(self.transport)
        except Exception as exc:  # noqa: BLE001
            tunnel.error = str(exc)
        self.tunnels.append(tunnel)
        return tunnel

    def stop(self, tunnel: Tunnel) -> None:
        tunnel.stop()
        if tunnel in self.tunnels:
            self.tunnels.remove(tunnel)

    def stop_all(self) -> None:
        for tunnel in list(self.tunnels):
            self.stop(tunnel)