如何用Python实现SSH远程端口转发（替代SSH客户端）

需求：让未安装SSH客户端的Windows用户，通过Python脚本将本地3000端口暴露到远程服务器的3001端口（允许所有外部IP访问），对应SSH命令：

ssh -R 0.0.0.0:3001:127.0.0.1:3000 -N root@SERVER_IP

原代码使用paramiko的本地端口转发逻辑无法满足需求，以下是修改后的远程端口转发实现：

import paramiko
import sys
import socket
import threading

def handle_data_forward(src, dst):
    """双向转发数据"""
    try:
        while True:
            data = src.recv(1024)
            if not data:
                break
            dst.send(data)
    except Exception as e:
        pass
    finally:
        src.close()
        dst.close()

def start_remote_port_forward(transport, remote_listen_host, remote_listen_port, local_target_host, local_target_port):
    """启动远程端口转发：远程端口流量转发到本地服务"""
    # 在远程服务器上请求监听指定端口（0.0.0.0允许所有IP访问）
    transport.request_port_forward(remote_listen_host, remote_listen_port)
    print(f"✅ 远程端口转发已启动：远程{remote_listen_host}:{remote_listen_port} → 本地{local_target_host}:{local_target_port}")
    print("按 Ctrl+C 停止转发")

    try:
        while True:
            # 接受远程服务器上的连接请求
            chan = transport.accept(timeout=10)
            if not chan:
                continue
            
            # 建立到本地服务的连接
            local_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            try:
                local_sock.connect((local_target_host, local_target_port))
            except Exception as e:
                print(f"❌ 无法连接本地服务：{e}")
                chan.close()
                continue
            
            # 启动双向数据转发线程
            threading.Thread(target=handle_data_forward, args=(chan, local_sock), daemon=True).start()
            threading.Thread(target=handle_data_forward, args=(local_sock, chan), daemon=True).start()
    except KeyboardInterrupt:
        print("\n🔴 端口转发已停止")
        transport.close()
        sys.exit(0)

if __name__ == "__main__":
    # 配置参数
    CONFIG = {
        "ssh_host": "SERVER_IP",          # 替换为你的SSH服务器IP
        "ssh_port": 22,                   # SSH服务端口
        "ssh_user": "root",               # SSH用户名
        "ssh_password": "pAss@123A",      # SSH密码
        "remote_listen_host": "0.0.0.0",  # 远程监听所有IP
        "remote_listen_port": 3001,       # 远程监听端口
        "local_target_host": "127.0.0.1", # 本地服务地址
        "local_target_port": 3000         # 本地服务端口
    }

    try:
        # 创建SSH传输连接
        transport = paramiko.Transport((CONFIG["ssh_host"], CONFIG["ssh_port"]))
        transport.connect(
            username=CONFIG["ssh_user"],
            password=CONFIG["ssh_password"],
            hostkey=None
        )

        # 启动远程端口转发
        start_remote_port_forward(
            transport,
            CONFIG["remote_listen_host"],
            CONFIG["remote_listen_port"],
            CONFIG["local_target_host"],
            CONFIG["local_target_port"]
        )
    except Exception as e:
        print(f"❌ 初始化失败：{e}")
        sys.exit(1)

关键说明 #

• 核心逻辑：通过paramiko.Transport.request_port_forward在远程服务器注册端口监听，再通过transport.accept接收远程连接，将流量转发到本地服务
• 权限配置：确保远程SSH服务器的sshd_config中GatewayPorts设置为yes或clientspecified（默认no会限制仅本地访问远程端口）
• 防火墙设置：远程服务器防火墙需开放3001端口的入站规则

内容的提问来源于stack exchange，提问作者Jeeva