SSH认证失败致代码无法执行的问题排查求助

我正在开发一套程序，功能是通过TCP服务器接收客户端命令，然后在SSH服务器上执行。运行时出现SSH认证失败的错误，已确认权限无问题，也尝试修改过IP地址，但问题依旧。

服务端代码 #

import socket
import paramiko
import threading

# SSH sunucu bilgileri
SSH_HOST = ("localhost")
SSH_PORT = 22
SSH_USERNAME = "skonuvs"
SSH_PRIVATE_KEY_PATH = "/home/skonuvs/rsakeys"
SSH_PASS = "lelele"


# TCP sunucu bilgileri
TCP_HOST = "0.0.0.0"
TCP_PORT = 53953

def ssh_connect(command):
    # SSH istemcisi oluştur
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

    try:
        # Özel anahtarı yükle
        ssh_private_key = paramiko.RSAKey.from_private_key_file(SSH_PRIVATE_KEY_PATH)

        # Sunucuya bağlan
        client.connect(SSH_HOST, port=SSH_PORT, username=SSH_USERNAME,password=SSH_PASS, pkey=ssh_private_key)
        print(f"Connected to {SSH_HOST}")

        # Komut çalıştırma
        stdin, stdout, stderr = client.exec_command(command)
        output = stdout.read() + stderr.read()
        client.send(output.encode())
        return output

    except Exception as e:
        print(f"Failed to connect to {SSH_HOST}: {e}")
        return str(e).encode()  # Hata durumunda dönecek veriyi encode edin

    finally:
        # Bağlantıyı kapat
        client.close()

def handle_client(client_socket):
    print("Client handler started")
    while True:
        try:
            # İstemciden veri al
            command = client_socket.recv(1024).decode()
            if not command:
                break

            print(f"Executing command: {command}")
            output = ssh_connect(command)
            client_socket.send(output)

        except Exception as e:
            print(f"Error handling client: {e}")
            break

    # Bağlantıları kapat
    client_socket.close()

def start_server():
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.bind((TCP_HOST, TCP_PORT))
    server.listen(5)
    print(f"Listening on {TCP_HOST}:{TCP_PORT}")

    while True:
        client_socket, addr = server.accept()
        print(f"Accepted connection from {addr}")
        client_handler = threading.Thread(target=handle_client, args=(client_socket,))
        client_handler.start()

if __name__ == "__main__":
    start_server()

客户端代码 #

import socket

# TCP sunucu bilgileri
TCP_HOST = "127.0.0.1"  # Sunucu adresini buraya yazın
TCP_PORT = 53953

def start_client():
    client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    client.connect((TCP_HOST, TCP_PORT))
    print(f"Connected to TCP server at {TCP_HOST}:{TCP_PORT}")

    while True:
        command = input("Enter command to execute on SSH server: ")
        if command.lower() == "exit":
            break

        # Komutu sunucuya gönder
        client.send(command.encode())

        # Sunucudan sonucu al
        response = client.recv(1024).decode()
        print(f"Response:\n{response}")

    # Bağlantıyı kapat
    client.close()

if __name__ == "__main__":
    start_client()

错误日志 #

Listening on 0.0.0.0:53953
Accepted connection from ('127.0.0.1', 42594)
Client handler started
Executing command: ls -la
Failed to connect to localhost: Authentication failed.

排查与修复 #

核心问题 #

1. 认证方式冲突：client.connect()同时传入password和pkey参数，Paramiko优先尝试密钥认证，若密钥认证失败，不会自动切换到密码认证，直接抛出认证失败错误。
2. 无效代码：client.send(output.encode())是错误用法，Paramiko的SSHClient对象没有send方法，这行代码无意义且可能引发额外错误。

修复步骤 #

方案1：使用密钥认证

• 移除client.connect()中的password=SSH_PASS参数。
• 如果私钥设置了密码，需要在加载密钥时传入密码：

  ssh_private_key = paramiko.RSAKey.from_private_key_file(SSH_PRIVATE_KEY_PATH, password=SSH_PASS)
  

• 先用命令行验证密钥有效性：ssh -i /home/skonuvs/rsakeys skonuvs@localhost，确保能正常连接。

方案2：使用密码认证

• 移除ssh_private_key相关代码，client.connect()仅保留密码参数：

  client.connect(SSH_HOST, port=SSH_PORT, username=SSH_USERNAME, password=SSH_PASS)
  

通用修复

删除ssh_connect函数中的client.send(output.encode())代码行，直接返回output即可。

修改后的服务端代码（密钥认证示例） #

import socket
import paramiko
import threading

# SSH 服务器信息
SSH_HOST = "localhost"
SSH_PORT = 22
SSH_USERNAME = "skonuvs"
SSH_PRIVATE_KEY_PATH = "/home/skonuvs/rsakeys"
# 若密钥有密码，取消注释并填写
# SSH_KEY_PASS = "lelele"

# TCP 服务器信息
TCP_HOST = "0.0.0.0"
TCP_PORT = 53953

def ssh_connect(command):
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

    try:
        # 加载私钥，有密码则添加password参数
        ssh_private_key = paramiko.RSAKey.from_private_key_file(SSH_PRIVATE_KEY_PATH)
        # ssh_private_key = paramiko.RSAKey.from_private_key_file(SSH_PRIVATE_KEY_PATH, password=SSH_KEY_PASS)

        # 仅用密钥认证连接
        client.connect(SSH_HOST, port=SSH_PORT, username=SSH_USERNAME, pkey=ssh_private_key)
        print(f"已连接到 {SSH_HOST}")

        stdin, stdout, stderr = client.exec_command(command)
        output = stdout.read() + stderr.read()
        return output

    except Exception as e:
        print(f"连接 {SSH_HOST} 失败: {e}")
        return str(e).encode()

    finally:
        client.close()

def handle_client(client_socket):
    print("客户端处理线程启动")
    while True:
        try:
            command = client_socket.recv(1024).decode()
            if not command:
                break

            print(f"正在执行命令: {command}")
            output = ssh_connect(command)
            client_socket.send(output)

        except Exception as e:
            print(f"处理客户端出错: {e}")
            break

    client_socket.close()

def start_server():
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.bind((TCP_HOST, TCP_PORT))
    server.listen(5)
    print(f"TCP服务器监听于 {TCP_HOST}:{TCP_PORT}")

    while True:
        client_socket, addr = server.accept()
        print(f"接受来自 {addr} 的连接")
        client_handler = threading.Thread(target=handle_client, args=(client_socket,))
        client_handler.start()

if __name__ == "__main__":
    start_server()

内容的提问来源于stack exchange，提问作者Muhammed Salih Arpacı