如何通过Python+Psycopg2+Paramiko连接AWS EC2上的PostgreSQL？

我是AWS EC2和PostgreSQL新手，已获取托管PostgreSQL的访问信息，且通过TablePlus成功连接数据库：

• 数据库详情：host=127.0.0.1、端口=5432、用户=test、数据库=test_data、SSL模式=prefer
• EC2服务器详情：地址=aa.bbb.ccc.dd、端口=23、用户=starlord

使用Python的psycopg2和paramiko库编写连接代码时，报错：Error connecting to the PostgreSQL database: encountered RSA key, expected OPENSSH key，求正确的连接及测试方法。

原配置文件 aws_db_config.json #

{
    "user": "test" (your_postgres_username),
    "password": "*********(1)" (your_postgres_password),
    "host": "aa.bbb.ccc.dd" (your_ec2_public_dns_or_ip),
    "port": "5432" (your_postgres_port),
    "database": "test_data",
    "sslmode": "prefer"  // or "disable" depending on your setup
}

原代码 #

import paramiko
import psycopg2
import json
import socket

def read_db_config_json(file_path):
    """Read database connection configuration from JSON file."""
    with open(file_path, 'r') as file:
        db_config = json.load(file)
    return db_config

def create_ssh_tunnel(ssh_host, ssh_port, ssh_user, ssh_key_path, local_port, remote_host, remote_port):
    """Create an SSH tunnel to the PostgreSQL server."""
    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    ssh.connect(ssh_host, port=ssh_port, username=ssh_user, key_filename=ssh_key_path)

    # Create a tunnel for the PostgreSQL connection
    transport = ssh.get_transport()
    tunnel = transport.open_channel(
        'direct-tcpip',
        (remote_host, remote_port),
        ('localhost', local_port)
    )
    return ssh, tunnel

def test_postgres_connection_via_ssh(config_file_path, ssh_config):
    """Connect to the PostgreSQL server via SSH tunneling and test the connection."""
    try:
        # Read the JSON config file
        db_config = read_db_config_json(config_file_path)

        # Create an SSH tunnel
        ssh, tunnel = create_ssh_tunnel(
            ssh_config['ssh_host'],
            ssh_config['ssh_port'],
            ssh_config['ssh_user'],
            ssh_config['ssh_key_path'],
            ssh_config['local_port'],
            db_config['host'],
            db_config['port']
        )

        # Update database config to use the local port
        db_config['host'] = 'localhost'
        db_config['port'] = ssh_config['local_port']

        # Establish the connection to PostgreSQL through the SSH tunnel
        connection = psycopg2.connect(**db_config)

        # Create a cursor object to execute SQL queries
        cursor = connection.cursor()

        # Query to get PostgreSQL version
        cursor.execute("SELECT version();")
        db_version = cursor.fetchone()
        print(f"Connected to PostgreSQL server: {db_version}\n")

        # Close the cursor and connection
        cursor.close()
        connection.close()

        # Close the SSH tunnel
        tunnel.close()
        ssh.close()

    except Exception as error:
        print(f"Error connecting to the PostgreSQL database: {error}")

if __name__ == "__main__":
    # Provide the path to your aws_db_config.json
    # SSH configuration for the EC2 instance
    ssh_config = {
        'ssh_host': 'aa.bbb.ccc.dd',
        'ssh_port': 23,  # default SSH port
        'ssh_user': 'starlord',
        'ssh_key_path': r'./saved.pem',
        'local_port': 5432  # local port for the tunnel
    }
    test_postgres_connection_via_ssh('aws_db_config.json', ssh_config)

1. 修复密钥格式问题 #

报错原因是paramiko不支持传统RSA格式密钥，需将.pem文件转为OPENSSH格式：

• 打开终端执行命令：

ssh-keygen -p -m PEM -f ./saved.pem

• 按提示输入密钥密码（无密码直接回车），转换完成后原密钥文件即变为OPENSSH格式。

2. 修正配置文件错误 #

原JSON配置的注释格式不符合规范，且数据库host需填写内部地址，修改后：

{
    "user": "test",
    "password": "你的PostgreSQL真实密码",
    "host": "127.0.0.1",
    "port": 5432,
    "database": "test_data",
    "sslmode": "prefer"
}

说明：数据库host填127.0.0.1，因为是通过EC2隧道访问内部数据库，EC2地址仅作为SSH连接主机。

3. 修正代码逻辑 #

调整SSH隧道的目标主机参数，确保端口为整数类型，并优化资源释放逻辑，修改后的完整代码：

import paramiko
import psycopg2
import json

def read_db_config_json(file_path):
    """读取JSON配置文件中的数据库连接信息"""
    with open(file_path, 'r') as file:
        db_config = json.load(file)
    # 确保端口为整数类型
    db_config['port'] = int(db_config['port'])
    return db_config

def create_ssh_tunnel(ssh_host, ssh_port, ssh_user, ssh_key_path, local_port, remote_host, remote_port):
    """创建SSH隧道连接PostgreSQL服务器"""
    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    try:
        ssh.connect(
            ssh_host,
            port=int(ssh_port),
            username=ssh_user,
            key_filename=ssh_key_path,
            look_for_keys=True,
            allow_agent=False
        )
        # 创建端口转发隧道
        transport = ssh.get_transport()
        tunnel = transport.open_channel(
            'direct-tcpip',
            (remote_host, remote_port),
            ('localhost', local_port)
        )
        return ssh, tunnel
    except Exception as e:
        ssh.close()
        raise e

def test_postgres_connection_via_ssh(config_file_path, ssh_config):
    """通过SSH隧道连接PostgreSQL并测试"""
    ssh = None
    tunnel = None
    connection = None
    try:
        db_config = read_db_config_json(config_file_path)
        # 创建SSH隧道，目标主机为数据库内部地址127.0.0.1
        ssh, tunnel = create_ssh_tunnel(
            ssh_config['ssh_host'],
            ssh_config['ssh_port'],
            ssh_config['ssh_user'],
            ssh_config['ssh_key_path'],
            ssh_config['local_port'],
            '127.0.0.1',
            db_config['port']
        )
        # 构造PostgreSQL连接参数
        pg_conn_config = {
            'user': db_config['user'],
            'password': db_config['password'],
            'host': 'localhost',
            'port': ssh_config['local_port'],
            'database': db_config['database'],
            'sslmode': db_config['sslmode']
        }
        connection = psycopg2.connect(**pg_conn_config)
        cursor = connection.cursor()
        cursor.execute("SELECT version();")
        db_version = cursor.fetchone()
        print(f"成功连接PostgreSQL，版本信息：{db_version[0]}")
        cursor.close()
    except Exception as error:
        print(f"连接PostgreSQL失败：{error}")
    finally:
        # 确保所有资源正常关闭
        if connection:
            connection.close()
        if tunnel:
            tunnel.close()
        if ssh:
            ssh.close()

if __name__ == "__main__":
    ssh_config = {
        'ssh_host': 'aa.bbb.ccc.dd',
        'ssh_port': 23,
        'ssh_user': 'starlord',
        'ssh_key_path': './saved.pem',
        'local_port': 5432
    }
    test_postgres_connection_via_ssh('aws_db_config.json', ssh_config)

4. 测试步骤 #

1. 确保密钥文件权限正确（Linux/macOS执行chmod 600 saved.pem，Windows无需额外设置）
2. 替换配置文件中的密码为真实PostgreSQL密码
3. 运行修改后的Python代码，输出PostgreSQL版本信息即为连接成功

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