Python通过SSH隧道远程连接MySQL数据库

橘子海

发布时间：2026-10-16| 10人浏览过

原创

本文介绍如何在Python中通过SSH隧道安全连接远程MySQL数据库。当数据库限制本地访问且端口非标准时，利用sshtunnel和pymysql库建立加密通道，实现远程数据交互。文章包含环境安装、代码封装及防注入注意事项，适用于需要安全访问内网数据库的开发场景。

背景

若需访问部署在远程服务器上的MySQL数据库，而该数据库出于安全考虑配置为仅允许本地连接（即必须登录至服务器本身方可使用），且其他远程访问被禁止、端口也已修改，此时需借助SSH协议建立连接。该方式与Navicat中通过SSH隧道连接数据库的界面操作原理一致。

Navicat

连接数据库

安装支持库

在Python中连接MySQL前，需先安装pymysql库。

pip install pymysql

同时安装用于SSH隧道的sshtunnel库。

pip install sshtunnel    #当前最新 0.3.1版

建议安装最新版本的sshtunnel库，旧版本可能存在已知bug。

连接Mysql

基于SSH连接MySQL的具体实现可参考sshtunnel官方文档，其中提供了相关示例代码。

with SSHTunnelForwarder(
        ('192.168.1.1', 2222),
        ssh_password='123456',
        ssh_username='root',
        remote_bind_address=('127.0.0.1', 3306)) as server:
    print('SSH连接成功')
    conn = pymysql.connect(host='127.0.0.1',
                           port=server.local_bind_port,
                           user='root',
                           database='data',
                           charset='utf8')
    print('mysql数据库连接成功')
    cursor = conn.cursor()
    ...  #获取数据操作，此处省略
    cursor.close()
    conn.close()

自定义查询函数

可将上述连接逻辑封装为独立函数，以便在项目的其他模块中复用。

def mysql_ssh(sql,args=None):
    with SSHTunnelForwarder(
            ('192.168.1.1', 2222),
            ssh_password='123456',
            ssh_username='root',
            remote_bind_address=('127.0.0.1', 3306)) as server:
        print('SSH连接成功')
        conn = pymysql.connect(host='127.0.0.1',
                               port=server.local_bind_port,
                               user='root',
                               database='data',
                               charset='utf8')
        print('mysql数据库连接成功')
        cursor = conn.cursor()
        print('游标获取成功')
        try:
            print(f'执行查询语句：{sql}  参数：{args}')
            cursor.execute(sql,args)
            print('数据查询成功')
            conn.commit()
            print('事务提交成功')
            datas = cursor.fetchall()
            success = True
        except:
            print('数据查询失败')
            datas = None
            success = False
        print('正在关闭数据库连接')
        cursor.close()
        conn.close()
    return datas, success

注意点：