Python实现SSH隧道功能的示例代码

pip install paramiko

import paramiko
from sshtunnel import SSHTunnelForwarder
 
def create_ssh_tunnel(ssh_host, ssh_port, ssh_user, ssh_password, remote_bind_address, local_bind_address):
    # 创建SSH隧道
    server = SSHTunnelForwarder(
        (ssh_host, ssh_port),
        ssh_username=ssh_user,
        ssh_password=ssh_password,
        remote_bind_address=remote_bind_address,
        local_bind_address=local_bind_address
    )
    
    # 启动SSH隧道
    server.start()
    print(f"SSH tunnel established: {local_bind_address[1]} -> {remote_bind_address}")
    return server
 
if __name__ == "__main__":
    # SSH服务器信息
    ssh_host = 'your.ssh.server.com'
    ssh_port = 22
    ssh_user = 'your_username'
    ssh_password = 'your_password'
    
    # 远程服务信息
    remote_bind_address = ('localhost', 80)
    
    # 本地监听地址
    local_bind_address = ('0.0.0.0', 8080)
    
    # 创建SSH隧道
    tunnel = create_ssh_tunnel(ssh_host, ssh_port, ssh_user, ssh_password, remote_bind_address, local_bind_address)
    
    try:
        # 保持程序运行
        while True:
            pass
    except KeyboardInterrupt:
        # 关闭SSH隧道
        tunnel.stop()
        print("SSH tunnel closed.")

import paramiko
 
def create_remote_tunnel(ssh_host, ssh_port, ssh_user, ssh_password, remote_bind_address, local_bind_address):
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    client.connect(ssh_host, ssh_port, ssh_user, ssh_password)
    
    transport = client.get_transport()
    transport.request_port_forward('', remote_bind_address[1])
    
    def handler(chan, host, port):
        sock = socket.socket()
        try:
            sock.connect((host, port))
        except Exception as e:
            print('Forwarding request to %s:%d failed: %r' % (host, port, e))
            return
        print('Connected!  Tunnel open %r -> %r -> %r' % (chan.origin_addr, chan.getpeername(), (host, port)))
        while True:
            r, w, x = select.select([sock, chan], [], [])
            if sock in r:
                data = sock.recv(1024)
                if len(data) == 0:
                    break
                chan.send(data)
            if chan in r:
                data = chan.recv(1024)
                if len(data) == 0:
                    break
                sock.send(data)
        chan.close()
        sock.close()
        print('Tunnel closed from %r' % (chan.origin_addr,))
    
    for i in range(transport.server_mode.port):
        transport.open_channel('direct-tcpip', local_bind_address, remote_bind_address, handler)
    
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print("Closing connections...")
        transport.close()
        client.close()
 
if __name__ == "__main__":
    # SSH服务器信息
    ssh_host = 'your.ssh.server.com'
    ssh_port = 22
    ssh_user = 'your_username'
    ssh_password = 'your_password'
    
    # 远程服务信息
    remote_bind_address = ('0.0.0.0', 9000)
    
    # 本地监听地址
    local_bind_address = ('localhost', 8080)
    
    # 创建远程SSH隧道
    create_remote_tunnel(ssh_host, ssh_port, ssh_user, ssh_password, remote_bind_address, local_bind_address)

pip install paramiko sshtunnel

import pymysql
from sshtunnel import SSHTunnelForwarder
 
# SSH服务器信息
ssh_host = '192.168.1.100'
ssh_port = 22
ssh_username = 'user'
ssh_password = 'password'
 
# 目标数据库信息
db_host = '192.168.1.200'
db_port = 3306
db_user = 'db_user'
db_password = 'db_password'
db_name = 'database_name'
 
# 创建SSH隧道
with SSHTunnelForwarder(
    (ssh_host, ssh_port),
    ssh_username=ssh_username,
    ssh_password=ssh_password,
    remote_bind_address=(db_host, db_port)
) as tunnel:
    # 隧道建立成功后，本地端口被转发到目标数据库
    local_port = tunnel.local_bind_port
    
    # 连接数据库
    conn = pymysql.connect(
        host='127.0.0.1',
        port=local_port,
        user=db_user,
        password=db_password,
        database=db_name
    )
    
    try:
        with conn.cursor() as cursor:
            # 执行SQL查询
            sql = "SELECT * FROM your_table"
            cursor.execute(sql)
            results = cursor.fetchall()
            for row in results:
                print(row)
    finally:
        # 关闭数据库连接
        conn.close()

pip install paramiko

import paramiko
from sshtunnel import SSHTunnelForwarder
 
# SSH 服务器信息
ssh_host = 'your_ssh_server_ip'
ssh_port = 22
ssh_user = 'your_username'
ssh_password = 'your_password'
 
# 远程服务信息
remote_bind_address = ('127.0.0.1', 3306)  # 假设远程服务是 MySQL 数据库
 
# 本地绑定地址
local_bind_address = ('127.0.0.1', 10022)
 
# 创建 SSH 隧道
with SSHTunnelForwarder(
    (ssh_host, ssh_port),
    ssh_username=ssh_user,
    ssh_password=ssh_password,
    remote_bind_address=remote_bind_address,
    local_bind_address=local_bind_address
) as tunnel:
    print(f"SSH tunnel established: {local_bind_address} -> {remote_bind_address}")
    
    # 在这里执行需要通过隧道的操作，例如连接到数据库
    import pymysql
    
    db = pymysql.connect(
        host='127.0.0.1',
        port=tunnel.local_bind_port,
        user='db_user',
        password='db_password',
        database='db_name'
    )
    
    cursor = db.cursor()
    cursor.execute("SELECT * FROM your_table")
    results = cursor.fetchall()
    for row in results:
        print(row)
    
    db.close()
 
print("SSH tunnel closed")