快捷導(dǎo)航

Python利用SSH隧道實(shí)現(xiàn)數(shù)據(jù)庫(kù)訪(fǎng)問(wèn)

更新時(shí)間：2025年03月08日 11:50:08 作者：Java菜鳥(niǎo)在北京

這篇文章主要為大家詳細(xì)介紹了如何通過(guò)sshtunnel類(lèi)庫(kù)建立SSH隧道,再使用paramiko通過(guò)SSH來(lái)訪(fǎng)問(wèn)數(shù)據(jù)庫(kù),感興趣的小伙伴可以跟隨小編一起學(xué)習(xí)一下

本文介紹通過(guò)sshtunnel類(lèi)庫(kù)建立SSH隧道，使用paramiko通過(guò)SSH來(lái)訪(fǎng)問(wèn)數(shù)據(jù)庫(kù)。

實(shí)現(xiàn)了兩種建立SSH方式：公私鑰驗(yàn)證、密碼驗(yàn)證。

公私鑰可讀本地，也可讀取Aws S3上的私鑰文件。

本質(zhì)上就是在本機(jī)建立SSH隧道，然后將訪(fǎng)問(wèn)DB轉(zhuǎn)發(fā)到本地SSH內(nèi)去訪(fǎng)問(wèn)數(shù)據(jù)庫(kù)。

簡(jiǎn)單易懂，上代碼：

from sshtunnel import SSHTunnelForwarder
from sqlalchemy import create_engine, text
import paramiko
import io
import socket
 
#### 都換成你自己的
# SSH配置 
ssh_host = '' #主機(jī)
ssh_port = 22  #端口
ssh_user = 'ec2-user' #用戶(hù)名
ssh_password = ''  #密碼（如果是密碼驗(yàn)證）
ssh_key_path = r'C:\Users\Desktop\test_primi.pem' #私鑰本地地址（如果是公私鑰驗(yàn)證）
# 數(shù)據(jù)庫(kù)配置
database_user = '' #用戶(hù)名
database_password = '' #密碼
database_name = '' #數(shù)據(jù)庫(kù)名
database_host = '' #主機(jī)
database_port = 3306 #端口號(hào)
 
def get_available_port():
    """
    獲取可用的本地端口。
    """
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    try:
        sock.bind(('127.0.0.1', 0))
        port = sock.getsockname()[1]
    finally:
        sock.close()
    return port
 
try:
    #讀取S3上的私鑰
    # s3_client = get_s3_client()
    # response = s3_client.get_object(Bucket='桶名稱(chēng)', Key='私鑰在S3上的key')
    # private_key_content = response['Body'].read().decode('utf-8')
    #private_key = paramiko.RSAKey.from_private_key(io.StringIO(private_key_content))
    #讀取本地私鑰
    private_key = paramiko.RSAKey.from_private_key_file(ssh_key_path)
    local_port = get_available_port()
 
    # 創(chuàng)建 SSH 隧道
    tunnel= SSHTunnelForwarder(
            (ssh_host, ssh_port),
            ssh_username=ssh_user,
            ssh_pkey=private_key,
            #ssh_password=ssh_password,
            remote_bind_address=(database_host, database_port),
            local_bind_address=('127.0.0.1', local_port),
            host_pkey_directories=[]
    )
    try:
        # 啟動(dòng) SSH 隧道
        tunnel.start()
        # 連接數(shù)據(jù)庫(kù)
        engine = create_engine(
            f'mysql+pymysql://{database_user}:{database_password}@127.0.0.1:{local_port}/{database_name}')
        # 測(cè)試數(shù)據(jù)庫(kù)連接
        with engine.connect() as connection:
            result = connection.execute(text("SELECT count(*) from activity_logs"))
            for row in result:
                print(row)
    finally:
        # 關(guān)閉 SSH 隧道
        tunnel.stop()
except Exception as e:
    print(f"出現(xiàn)錯(cuò)誤: {e}")

執(zhí)行結(jié)果：

成功輸出條數(shù)

注意：host_pkey_directories=[] 的意思是不要在指定的目錄中尋找密鑰，如果沒(méi)有將出現(xiàn)如下錯(cuò)誤，但不影響程序正常執(zhí)行。

可以先在本地用Navicat之類(lèi)的客戶(hù)端測(cè)好了，sshtunnel應(yīng)該是三種驗(yàn)證方法都支持的，源碼如下

到此這篇關(guān)于Python利用SSH隧道實(shí)現(xiàn)數(shù)據(jù)庫(kù)訪(fǎng)問(wèn)的文章就介紹到這了,更多相關(guān)Python SSH訪(fǎng)問(wèn)數(shù)據(jù)庫(kù)內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: