无法使用Paramiko连接MySQL数据库，可以使用sshtunnel

Question

我的代码使用

sshtunnel

并正确连接到我的数据库。然而，由于

grequests

和

sshtunnel

之间的冲突，我不得不改用另一个库来实现

ssh

，并且在

Paramiko

上运气很好（因为它与

monkeypatch

上的

grequests

一起使用）。

ssh tunnel

部分工作正常，但是我现在收到消息：

An error occurred: 1045 (28000): Access denied for user 'PA_username'@'localhost' (using password: YES)

我使用的环境变量没有改变，据我所知，带有

sshtunnel

和不带

grequests

的工作代码以及无法连接的带有

Paramiko

的版本之间的一切都是相同的。

完整代码在这里：

from gevent import monkey

monkey.patch_all()

import grequests
import mysql.connector
import paramiko
from pprint import pprint
import os

ssh_password = os.environ.get("PA_SSH_PASS")
db_password = os.environ.get("PA_DB_PASS")
database = os.environ.get("PA_DB_NAME")

def get_current_listing_urls():
    try:
        ssh_client = paramiko.SSHClient()
        ssh_client.load_system_host_keys()
        ssh_client.set_missing_host_key_policy(paramiko.WarningPolicy)

        ssh_client.connect(
            "ssh.eu.pythonanywhere.com",
            username=PA_username,
            password=ssh_password,
        )

        transport = ssh_client.get_transport()

        channel = transport.open_channel(
            "direct-tcpip",
            dest_addr=("PA_username.mysql.eu.pythonanywhere-services.com", 3306),
            src_addr=("127.0.0.1", 3306),
        )

        db = mysql.connector.connect(
            user="PA_username",
            password=db_password,
            host="127.0.0.1",
            port=3306,
            database=database,
            use_pure=True,
        )

        table_name = "listings"

        cursor = db.cursor(dictionary=True)

        query = f"SELECT COUNT(*) FROM {table_name};"

        cursor.execute(query)

        results = cursor.fetchall()

        return results

    except Exception as e:
        print(f"An error occurred: {str(e)}")
    finally:
        if "cursor" in locals() and cursor:
            cursor.close()
        if "db" in locals() and db:
            db.close()
        if "ssh_client" in locals() and ssh_client:
            ssh_client.close()


old_listing_urls = get_current_listing_urls()

pprint(old_listing_urls)

我已经尝试了几乎所有我能想到的方法，但我看不出我哪里出错了。该消息可能意味着它正在连接并尝试进行身份验证，但失败。然而，数据库名称、用户名和密码都是正确的，因为它们在使用 sshtunnel 时都有效。

任何帮助将不胜感激！谢谢

编辑：

带有

sshtunnel

的工作代码在这里：

import mysql.connector
import sshtunnel
from pprint import pprint
import time
import os
import json

sshtunnel.SSH_TIMEOUT = 5.0
sshtunnel.TUNNEL_TIMEOUT = 5.0

ssh_password = os.environ.get("PA_SSH_PASS")
db_password = os.environ.get("PA_DB_PASS")
database = os.environ.get("PA_DB_NAME")


def get_current_listing_urls():
    try:
        with sshtunnel.SSHTunnelForwarder(
            ("ssh.eu.pythonanywhere.com"),
            ssh_username=PA_username,
            ssh_password=ssh_password,
            remote_bind_address=(
                "PA_username.mysql.eu.pythonanywhere-services.com",
                3306,
            ),
            threaded=False,
        ) as tunnel:
            db = mysql.connector.connect(
                user=PA_username,
                password=db_password,
                host="127.0.0.1",
                port=tunnel.local_bind_port,
                database=database,
                use_pure=True,
            )

            table_name = "listings"

            cursor = db.cursor(dictionary=True)

            query = f"SELECT COUNT(*)  FROM {table_name};"  

            cursor.execute(query)

            results = cursor.fetchall()

            return results

    except Exception as e:
        print(f"An error occurred: {str(e)}")
    finally:
        if "cursor" in locals() and cursor:
            cursor.close()
        if "db" in locals() and db:
            db.close()
        if "tunnel" in locals() and tunnel:
            tunnel.close()


old_listing_urls = get_current_listing_urls()

pprint(old_listing_urls)

Answer 1

我无法按照我想要的方式解决这个问题，最后我使用子进程在cmd中运行ssh命令并随后将其关闭。这需要 rsa 密钥而不是密码，因为发送 ssh 命令后我无法在 cmd 中输入密码。通常 pevent 将用于此目的，但所需的功能（spawn）在 Windows 上不起作用。使用密钥意味着无需输入密码。

完整代码：

import subprocess
from db_utilities import get_current_listing_urls

# Define the SSH command to establish a tunnel
ssh_command = "ssh -L 3333:PA_username.mysql.pythonanywhere-services.com:3306 [email protected]"

# Open a new cmd window and run the SSH command
cmd_process = subprocess.Popen(
    ["cmd", "/K", ssh_command],
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    universal_newlines=True,
    text=True,
)

try:
    results = get_current_listing_urls()
    print(results)
finally:
    # Check if the process is still running, and terminate it if necessary
    if cmd_process.poll() is None:
        print("Closing SSH Tunnel")
        cmd_process.terminate()