使用 paramiko 端口转发演示时如何避免无限循环？

Question

我需要在 Python 中使用端口转发通过 SSH 隧道与远程 MySQL DB 进行通信。我下载了 paramiko 包并尝试了端口转发演示 (forward.py)。它工作得很好，但我在将其集成到我自己的脚本中时遇到了麻烦（类似于下面的脚本）。当调用主转发函数时，它会进入无限循环，并且我的其余代码不会执行。如何使用forward.py 演示并摆脱无限循环？

我的脚本：

import paramiko
import forward
import MySQLdb
import cfg
import sys

client = paramiko.SSHClient()
client.load_system_host_keys()
client.set_missing_host_key_policy(paramiko.WarningPolicy())

try:
    client.connect(cfg.remhost, cfg.remport, username=cfg.user, password=cfg.password)
except Exception, e:
    print '*** Failed to connect to %s:%d: %r' % (cfg.remhost, cfg.remport, e)
    sys.exit(1)


try:
    forward.forward_tunnel(3306, cfg.remhost, 3306, client.get_transport())
except KeyboardInterrupt:
    print 'C-c: Port forwarding stopped.'
    sys.exit(0)

try:
    db = MySQLdb.connect('127.0.0.1', cfg.dbuser, cfg.dbpass, cfg.dbname)
except Exception, e:
    print 'Failed to connect to database'
    sys.exit(1)

try:
    cursor = self.db.cursor(MySQLdb.cursors.DictCursor)
    sql = 'SELECT * FROM  ' + cfg.dbtable
    cursor.execute(sql)
    results = cursor.fetchall()
    print str(len(results))
except Exception, e:
    print 'Failed to query database'
    sys.exit(1)

这是forward.py演示代码的主要部分：

class ForwardServer (SocketServer.ThreadingTCPServer):
    daemon_threads = True
    allow_reuse_address = True


class Handler (SocketServer.BaseRequestHandler):

    def handle(self):
        try:
            chan = self.ssh_transport.open_channel('direct-tcpip',
                                                   (self.chain_host, self.chain_port),
                                                   self.request.getpeername())
        except Exception, e:
            verbose('Incoming request to %s:%d failed: %s' % (self.chain_host,
                                                              self.chain_port,
                                                              repr(e)))
            return
        if chan is None:
            verbose('Incoming request to %s:%d was rejected by the SSH server.' %
                    (self.chain_host, self.chain_port))
            return

        verbose('Connected!  Tunnel open %r -> %r -> %r' % (self.request.getpeername(),
                                                            chan.getpeername(), (self.chain_host, self.chain_port)))
        while True:
            r, w, x = select.select([self.request, chan], [], [])
            if self.request in r:
                data = self.request.recv(1024)
                if len(data) == 0:
                    break
                chan.send(data)
            if chan in r:
                data = chan.recv(1024)
                if len(data) == 0:
                    break
                self.request.send(data)
        chan.close()
        self.request.close()
        verbose('Tunnel closed from %r' % (self.request.getpeername(),))


def forward_tunnel(local_port, remote_host, remote_port, transport):
    # this is a little convoluted, but lets me configure things for the Handler
    # object.  (SocketServer doesn't give Handlers any way to access the outer
    # server normally.)
    class SubHander (Handler):
        chain_host = remote_host
        chain_port = remote_port
        ssh_transport = transport
    ForwardServer(('', local_port), SubHander).serve_forever()


def verbose(s):
    if g_verbose:
        print s

Answer 1

我认为您需要处理程序转发代码在其自己的线程中运行并使用队列与其通信。

或者拉出相关的调用并将其放入您自己的循环中。嗯，我不确定这到底是如何工作的。

您希望 mysql 调用在您的程序中使用它还是从其他程序转发？

Answer 2

我是一个名为 Python X2Go 的 Pyhon 模块的上游作者，该模块大量使用 Paramiko 进行 SSH 会话管理。

请看一下forward.py文件中的代码：
http://code.x2go .org/gitweb?p=python-x2go.git;a=blob;f=x2go/forward.py

Python X2Go 的代码大量使用 Python gevent 进行服务器客户端通信、端口转发等。

问候，
麦克风