Twisted Conch SSHCommandClientEndpoint SSH密钥认证失败问题

你遇到的问题是使用SSHCommandClientEndpoint时，密码认证正常但密钥认证失败，且手动使用该密钥可正常登录目标机器。以下是排查和解决的关键点：

1. 检查私钥加载是否正确 #

情况1：私钥有密码保护 #

如果你的id_rsa是带密码的，直接用keys.Key.fromFile加载会失败，需要显式传入密码：

key = keys.Key.fromFile("id_rsa", passphrase=b"你的私钥密码")

情况2：私钥路径问题 #

确保程序运行时能找到id_rsa文件，建议使用绝对路径：

import os
key_path = os.path.abspath("./id_rsa")
key = keys.Key.fromFile(key_path)

2. 确认Endpoint参数格式 #

SSHCommandClientEndpoint.newConnection的host参数需要是字符串（IP或域名），不要用元组或其他格式：

endpoint = SSHCommandClientEndpoint.newConnection(
    reactor, 
    "ls", 
    "user", 
    "目标主机IP/域名",  # 这里直接传字符串
    22, 
    keys=[key]
)

3. 开启调试日志定位具体错误 #

添加日志可以看到认证过程中的详细报错，帮助定位问题：

import logging
logging.basicConfig(level=logging.DEBUG)
from twisted.python import log
log.startLogging(open('twisted_ssh.log', 'w'))

运行后查看日志文件，会显示认证失败的具体原因（比如密钥不匹配、服务器拒绝密钥等）。

4. 强制指定密钥认证方法（可选） #

如果Twisted默认的认证顺序导致问题，可以自定义认证客户端，强制优先使用密钥认证：

from twisted.conch.ssh.userauth import SSHUserAuthClient

class KeyOnlyAuth(SSHUserAuthClient):
    def getPassword(self, prompt=None):
        # 跳过密码认证，只尝试密钥
        return defer.fail(NotImplementedError())

# 修改Endpoint的创建方式，指定authClientFactory
endpoint = SSHCommandClientEndpoint.newConnection(
    reactor, 
    "ls", 
    "user", 
    "目标主机IP/域名", 
    22, 
    keys=[key],
    authClientFactory=lambda user, options: KeyOnlyAuth(user, options)
)

修改后的完整示例代码 #

from twisted.internet import protocol, defer, reactor, task
from twisted.conch.endpoints import SSHCommandClientEndpoint
from twisted.conch.ssh import keys
import os
import logging
from twisted.python import log

# 开启调试日志
logging.basicConfig(level=logging.DEBUG)
log.startLogging(open('twisted_ssh.log', 'w'))

async def conchSSH():
    finished = defer.Deferred()

    class ShowOutput(protocol.Protocol):
        received = b""
        def dataReceived(self, data):
            self.received += data
        def connectionLost(self, reason):
            finished.callback(self.received)

    # 用绝对路径加载私钥，若有密码需加passphrase参数
    key_path = os.path.abspath("id_rsa")
    key = keys.Key.fromFile(key_path)  # 若有密码：keys.Key.fromFile(key_path, passphrase=b"your_pass")

    endpoint = SSHCommandClientEndpoint.newConnection(
        reactor, 
        "ls", 
        "user", 
        "目标主机IP/域名", 
        22, 
        keys=[key]
    )
    factory = protocol.Factory.forProtocol(ShowOutput)
    await endpoint.connect(factory)
    print("SSH response:", await finished)

task.react(lambda *a, **k: defer.ensureDeferred(conchSSH()))

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