Python Twisted Conch - 如何通过多个连接停止 react 器？

Question

让我从这个开始……我根本不懂 Python；我在兜圈子，我根本不明白。我对替代和更简单的方法完全开放。

我的目标 :连接到不同的服务器，在每个服务器上运行相同的命令，然后(如现在/还没有)将输出用于生产性事情。惊人的。

我有什么 :在某处找到了一些代码(我会尝试找到一个链接并更新它)。我稍微修改了一下。它连接到不同的服务器，运行相同的命令。

问题 : 一切都完成后，我不知道如何停止 react 堆。我真的很想停止它而不按cntrl+c .我想我需要推迟一些事情，但我不知道什么或在哪里。我觉得当 SSHChannel 关闭时，需要以某种方式冒泡到 SSHConnection，以停止服务......所以传输可以知道发生了什么？我一直想以某种方式包装每个 reactor.connectTCP(server, 22, factory)以某种方式推迟。而且我觉得我可能需要一个 Controller 类。我尝试了这些东西，但没有正确尝试。也许gatherResults可能会有所帮助，但是，我不知道到底该放什么。

from twisted.conch.ssh import transport, connection, userauth, channel, common
from twisted.internet import defer, protocol, reactor
import sys, struct  

USER = 'username'
PASS = 'thisisforpersonalusesoicanstoreit!' 
CMD  = 'echo "merely this and nothing more"'


from twisted.python import log
import sys
log.startLogging(sys.stdout)


class ClientCommandTransport(transport.SSHClientTransport):
    def __init__(self, username, password, command):
        self.username = username
        self.password = password
        self.command  = command

    def verifyHostKey(self, pubKey, fingerprint):
        print fingerprint 
        return defer.succeed(True)

    def connectionSecure(self):
        self.requestService(
            PasswordAuth(self.username, self.password,
                         ClientConnection(self.command)))    

class PasswordAuth(userauth.SSHUserAuthClient):
    def __init__(self, user, password, connection):
        userauth.SSHUserAuthClient.__init__(self, user, connection)
        self.password = password

    def getPassword(self, prompt=None):
        return defer.succeed(self.password)

class ClientConnection(connection.SSHConnection):
    def __init__(self, cmd, *args, **kwargs):
        connection.SSHConnection.__init__(self)
        self.command = cmd

    def serviceStarted(self):
        self.openChannel(CommandChannel(self.command, conn=self))  

class CommandChannel(channel.SSHChannel):
    name = 'session'

    def __init__(self, command, *args, **kwargs):
        channel.SSHChannel.__init__(self, *args, **kwargs)
        self.command = command
        self.data = ''

    def channelOpen(self, data):
        self.conn.sendRequest(
            self, 'exec', common.NS(self.command), wantReply=True).addCallback(
                                                            self._gotResponse)

    def _gotResponse(self, _):
        self.conn.sendEOF(self) 
        self.loseConnection() 

    def dataReceived(self, data):
        #self.data += data
        print data 

    def request_exit_status(self, data):
        (status,) = struct.unpack('>L', data)
        # print 'exit status = ', status  

class ClientCommandFactory(protocol.ClientFactory):
    def __init__(self, command=CMD):
        self.username = USER
        self.password = PASS
        self.command  = command

    def buildProtocol(self, addr):
        protocol = ClientCommandTransport(
            self.username, self.password, self.command)
        return protocol    


masters = ['server1','server2','server3','server4','server5']

factory = ClientCommandFactory()

for server in masters:
    print server
    reactor.connectTCP(server, 22, factory)

reactor.run()

我确实玩过推迟 getPage对于 http 请求(确实有效)，但我似乎无法使用 react 器和 ssh 连接重新应用它。

这些是我真正希望我能理解的资源:

http://twistedmatrix.com/documents/current/api/twisted.internet.defer.gatherResults.html

http://mumak.net/stuff/twisted-disconnect.html

Python Twisted Stopping The Reactor With Multiple Clients

Best way to run remote commands thru ssh in Twisted?

What is the correct way to close a Twisted conch SSH connection?

以及所有扭曲的类定义..

使用下面的一个答案...我测试了传递对工厂的引用并最终在 SSHChannel 中停止了校长。 closed()如果工厂在其数组(或任何 python 调用的数组)中不再有连接。

我更新了工厂，现在也包括这个方法:

class ClientCommandFactory(protocol.ClientFactory): 

    def clientConnectionLost(self, connector, reason):
        print reason

我看了一下日志记录，因为我通常对正在发生的事情感兴趣并且......(其中一些是我自己的陈述，一些是默认的)

014-10-16 13:42:58-0500 [SSHChannel session (0) on SSHService ssh-connection on ClientCommandTransport,client] closed last TCP connection
2014-10-16 13:42:58-0500 [ClientCommandTransport,client] service stopped 
2014-10-16 13:42:58-0500 [ClientCommandTransport,client] connection lost
2014-10-16 13:42:58-0500 [ClientCommandTransport,client] [Failure instance: Traceback (failure with no frames): <class 'twisted.internet.error.ConnectionLost'>: Connection to the other side was lost in a non-clean fashion: Connection lost.
2014-10-16 13:42:58-0500 [ClientCommandTransport,client] ]
2014-10-16 13:42:58-0500 [ClientCommandTransport,client] connection lost
2014-10-16 13:42:58-0500 [ClientCommandTransport,client] [Failure instance: Traceback (failure with no frames): <class 'twisted.internet.error.ConnectionLost'>: Connection to the other side was lost in a non-clean fashion: Connection lost.
2014-10-16 13:42:58-0500 [ClientCommandTransport,client] ]
2014-10-16 13:42:58-0500 [ClientCommandTransport,client] Stopping factory <__main__.ClientCommandFactory instance at 0x02323030>
2014-10-16 13:42:58-0500 [-] Main loop terminated.

所以......它说连接以不干净的方式丢失。有没有更好的方法让我停下来..？

Answer 1

所以首先这是行不通的，因为connectTCP接受带有 IP address 的字符串作为第一个参数，您正在传递此列表中的元素:

masters = ['server1','server2','server3','server4','server5']

在所有任务完成后停止 react 器是扭曲的非常常见的用例。一种方法是存储要在工厂中执行的任务计数器。每次实例化该工厂的协议(protocol)实例时，将该数字加一，每次协议(protocol)实例(任务)返回结果时，减少计数器，当计数器达到 0 时停止 react 器。示例代码:

from twisted.conch.ssh import transport, connection, userauth, channel, common
from twisted.internet import defer, protocol, reactor
import sys, struct  

USER = 'username'
PASS = 'thisisforpersonalusesoicanstoreit!' 
CMD  = 'echo "merely this and nothing more"'


from twisted.python import log
import sys
log.startLogging(sys.stdout)


class ClientCommandTransport(transport.SSHClientTransport):
    def __init__(self, username, password, command, factory):
        self.username = username
        self.password = password
        self.command  = command
        self.factory = factory

    def verifyHostKey(self, pubKey, fingerprint):
        print fingerprint 
        return defer.succeed(True)

    def connectionSecure(self):
        self.requestService(
            PasswordAuth(self.username, self.password,
                         ClientConnection(self.command, self.factory)))

class PasswordAuth(userauth.SSHUserAuthClient):
    def __init__(self, user, password, connection):
        userauth.SSHUserAuthClient.__init__(self, user, connection)
        self.password = password

    def getPassword(self, prompt=None):
        return defer.succeed(self.password)

class ClientConnection(connection.SSHConnection):
    def __init__(self, cmd, *args, **kwargs):
        connection.SSHConnection.__init__(self)
        self.command = cmd
        self.factory = factory

    def serviceStarted(self):
        self.openChannel(CommandChannel(self.command, self.factory, conn=self))

class CommandChannel(channel.SSHChannel):
    name = 'session'

    def __init__(self, command, factory, *args, **kwargs):
        channel.SSHChannel.__init__(self, *args, **kwargs)
        self.command = command
        self.data = ''
        self.factory = factory
        self.factory.num_connections += 1
        self.factory.connections.append(self)

    def channelOpen(self, data):
        self.conn.sendRequest(
            self, 'exec', common.NS(self.command), wantReply=True).addCallback(
                                                            self._gotResponse)

    def _gotResponse(self, _):
        self.conn.sendEOF(self) 
        self.loseConnection()
        self.factory.num_connections -= 1
        self.factory.connections.remove(self)
        if self.factory.num_connections == 0:
            reactor.stop()

    def dataReceived(self, data):
        #self.data += data
        print data 

    def request_exit_status(self, data):
        (status,) = struct.unpack('>L', data)
        # print 'exit status = ', status  

class ClientCommandFactory(protocol.ClientFactory):
    def __init__(self, command=CMD):
        self.username = USER
        self.password = PASS
        self.command  = command
        self.connections = []
        self.num_connections = 0

    def buildProtocol(self, addr):
        protocol = ClientCommandTransport(
            self.username, self.password, self.command, self)
        return protocol    


masters = ['server1','server2','server3','server4','server5']

factory = ClientCommandFactory()

for server in masters:
    print server
    reactor.connectTCP(server, 22, factory)

reactor.run()

我在这里所做的是向工厂添加了两个变量 self.connections和 self.num_connections存储对工厂中连接的引用并计算连接数。然后在工厂的 buildProtocol工厂将自己传递给 ClientCommandTransport ，这又将工厂的引用传递给 ClientConnection ，它最终将引用传递给需要它的工厂 - 到 CommandChannel .每次 CommandChannel 的实例被实例化了，它有一个对工厂的引用，所以它把连接数加一并将自己添加到连接列表中，连接列表存储在工厂中。我假设 _gotResponse完成任务/命令时触发回调。因此，每当它被触发时，它都会像以前一样失去连接，但现在，它还会减少连接计数器并从工厂中删除对自身的引用。它还检查是否有任何其他打开的连接，如果没有，它会停止 react 器。

我没有测试过这段代码，但它是 Twisted 中的一种常见模式，工厂保留对其创建的协议(protocol)实例的引用列表，以便每个实例都可以通过工厂访问其他实例并且能够停止 react 器一次所有实例都完成了他们的工作。

请注意，此层次结构也有些深， Factory -> ClientCommandTransport -> ClientConnection -> CommandChannel而且我不确定将引用一直传递给工厂是否是最佳解决方案。

其中一个变量实际上是多余的——您只能存储 self.num_connections并增加/减少它或 self.connections ，从列表中添加/删除实例并使用 len(self.connections)看看是否还有任何打开的连接。