后续 - 令人高兴的是，我在下面引用的故障单现已解决.更简单的 API 将包含在 Twisted 的下一个版本中.原始答案仍然是使用 Conch 的有效方法，并且可能会揭示一些有关正在发生的事情的有趣细节，但是从 Twisted 13.1 开始，如果您只想运行命令并处理它的 I/O，这个更简单的界面可以使用.

Followup - Happily, the ticket I referenced below is now resolved. The simpler API will be included in the next release of Twisted. The original answer is still a valid way to use Conch and may reveal some interesting details about what's going on, but from Twisted 13.1 and on, if you just want to run a command and handle it's I/O, this simpler interface will work.

不幸的是，使用 Conch 客户端 API 在 SSH 上执行命令需要大量代码.Conch 使您可以处理许多不同的层，即使您只想要明智而无聊的默认行为.然而，这当然是可能的.这是我一直想完成并添加到 Twisted 以简化这种情况的一些代码:

It takes an unfortunately large amount of code to execute a command on an SSH using the Conch client APIs. Conch makes you deal with a lot of different layers, even if you just want sensible boring default behavior. However, it's certainly possible. Here's some code which I've been meaning to finish and add to Twisted to simplify this case:

import sys, os

from zope.interface import implements

from twisted.python.failure import Failure
from twisted.python.log import err
from twisted.internet.error import ConnectionDone
from twisted.internet.defer import Deferred, succeed, setDebugging
from twisted.internet.interfaces import IStreamClientEndpoint
from twisted.internet.protocol import Factory, Protocol

from twisted.conch.ssh.common import NS
from twisted.conch.ssh.channel import SSHChannel
from twisted.conch.ssh.transport import SSHClientTransport
from twisted.conch.ssh.connection import SSHConnection
from twisted.conch.client.default import SSHUserAuthClient
from twisted.conch.client.options import ConchOptions

# setDebugging(True)


class _CommandTransport(SSHClientTransport):
    _secured = False

    def verifyHostKey(self, hostKey, fingerprint):
        return succeed(True)


    def connectionSecure(self):
        self._secured = True
        command = _CommandConnection(
            self.factory.command,
            self.factory.commandProtocolFactory,
            self.factory.commandConnected)
        userauth = SSHUserAuthClient(
            os.environ['USER'], ConchOptions(), command)
        self.requestService(userauth)


    def connectionLost(self, reason):
        if not self._secured:
            self.factory.commandConnected.errback(reason)



class _CommandConnection(SSHConnection):
    def __init__(self, command, protocolFactory, commandConnected):
        SSHConnection.__init__(self)
        self._command = command
        self._protocolFactory = protocolFactory
        self._commandConnected = commandConnected


    def serviceStarted(self):
        channel = _CommandChannel(
            self._command, self._protocolFactory, self._commandConnected)
        self.openChannel(channel)



class _CommandChannel(SSHChannel):
    name = 'session'

    def __init__(self, command, protocolFactory, commandConnected):
        SSHChannel.__init__(self)
        self._command = command
        self._protocolFactory = protocolFactory
        self._commandConnected = commandConnected


    def openFailed(self, reason):
        self._commandConnected.errback(reason)


    def channelOpen(self, ignored):
        self.conn.sendRequest(self, 'exec', NS(self._command))
        self._protocol = self._protocolFactory.buildProtocol(None)
        self._protocol.makeConnection(self)


    def dataReceived(self, bytes):
        self._protocol.dataReceived(bytes)


    def closed(self):
        self._protocol.connectionLost(
            Failure(ConnectionDone("ssh channel closed")))



class SSHCommandClientEndpoint(object):
    implements(IStreamClientEndpoint)

    def __init__(self, command, sshServer):
        self._command = command
        self._sshServer = sshServer


    def connect(self, protocolFactory):
        factory = Factory()
        factory.protocol = _CommandTransport
        factory.command = self._command
        factory.commandProtocolFactory = protocolFactory
        factory.commandConnected = Deferred()

        d = self._sshServer.connect(factory)
        d.addErrback(factory.commandConnected.errback)

        return factory.commandConnected



class StdoutEcho(Protocol):
    def dataReceived(self, bytes):
        sys.stdout.write(bytes)
        sys.stdout.flush()


    def connectionLost(self, reason):
        self.factory.finished.callback(None)



def copyToStdout(endpoint):
    echoFactory = Factory()
    echoFactory.protocol = StdoutEcho
    echoFactory.finished = Deferred()
    d = endpoint.connect(echoFactory)
    d.addErrback(echoFactory.finished.errback)
    return echoFactory.finished



def main():
    from twisted.python.log import startLogging
    from twisted.internet import reactor
    from twisted.internet.endpoints import TCP4ClientEndpoint

    # startLogging(sys.stdout)

    sshServer = TCP4ClientEndpoint(reactor, "localhost", 22)
    commandEndpoint = SSHCommandClientEndpoint("/bin/ls", sshServer)

    d = copyToStdout(commandEndpoint)
    d.addErrback(err, "ssh command / copy to stdout failed")
    d.addCallback(lambda ignored: reactor.stop())
    reactor.run()



if __name__ == '__main__':
    main()

注意事项:

• 它使用 Twisted 10.1 中引入的新端点 API.可以直接在 reactor.connectTCP 上执行此操作，但我将其作为端点来执行以使其更有用；无需知道实际请求连接的代码即可轻松交换端点.
• 它根本不进行主机密钥验证！_CommandTransport.verifyHostKey 是您实现它的地方.查看 twisted/conch/client/default.py 以获取有关您可能想要做的事情的一些提示.
• 需要 $USER 作为远程用户名，您可能希望将其作为参数.
• 它可能仅适用于密钥身份验证.如果您想启用密码身份验证，您可能需要子类化 SSHUserAuthClient 并覆盖 getPassword 以执行某些操作.
• 几乎所有 SSH 和 Conch 层都在这里可见:
  • _CommandTransport 位于底部，这是一个实现 SSH 传输协议的普通旧协议.它创建了一个...
  • _CommandConnection 实现协议的 SSH 连接协商部分.一旦完成，一个...
  • _CommandChannel 用于与新打开的 SSH 通道通信._CommandChannel 执行实际的 exec 来启动您的命令.一旦通道打开，它就会创建一个...的实例
  • StdoutEcho，或您提供的任何其他协议.该协议将从您执行的命令中获取输出，并可以写入命令的标准输入.
  • It uses the new endpoint APIs introduced in Twisted 10.1. It's possible to do this directly on reactor.connectTCP, but I did it as an endpoint to make it more useful; endpoints can be swapped easily without the code that actually asks for a connection knowing.
  • It does no host key verification at all! _CommandTransport.verifyHostKey is where you would implement that. Take a look at twisted/conch/client/default.py for some hints about what kinds of things you might want to do.
  • It takes $USER to be the remote username, which you may want to be a parameter.
  • It probably only works with key authentication. If you want to enable password authentication, you probably need to subclass SSHUserAuthClient and override getPassword to do something.
  • Almost all of the layers of SSH and Conch are visible here:
    • _CommandTransport is at the bottom, a plain old protocol that implements the SSH transport protocol. It creates a...
    • _CommandConnection which implements the SSH connection negotiation parts of the protocol. Once that completes, a...
    • _CommandChannel is used to talk to a newly opened SSH channel. _CommandChannel does the actual exec to launch your command. Once the channel is opened it creates an instance of...
    • StdoutEcho, or whatever other protocol you supply. This protocol will get the output from the command you execute, and can write to the command's stdin.

    请参阅 http://twistedmatrix.com/trac/ticket/4698 了解在纠结于用更少的代码支持这个.

    See http://twistedmatrix.com/trac/ticket/4698 for progress in Twisted on supporting this with less code.