Paramiko Sessions 关闭子进程中的传输

Question

我们正在使用 paramiko 来创建一个连接库，该库大量使用其 get_pty 或 invoke_shell 功能。我们的库使用这些通道与目标设备进行交互。

但是每当我们使用multiprocessing 库时，我们都无法在子进程中使用 paramiko 连接句柄。 transport 在子进程中被关闭。

Is there a way to tell paramiko not to close the connection/channel at fork. 

这是重现问题的示例程序

from paramiko import SSHClient, AutoAddPolicy
from multiprocessing import Process
import logging
log = logging.getLogger("paramiko.transport").setLevel(1)

client = SSHClient()

client.set_missing_host_key_policy(AutoAddPolicy())

client.connect(hostname="localhost")

def simple_work(handle):
    print("==== ENTERED CHILD PROCESS =====")
    stdin, stdout, stderr = handle.exec_command("ifconfig")
    print(stdout.read())
    print("==== EXITING CHILD PROCESS =====")

p = Process(target=simple_work, args=(client,))
p.start()
p.join(2)
print("==== MAIN PROCESS AFTER JOIN =====")
stdin, stdout, stderr = client.exec_command("ls")
print(stdout.read())

这是错误

==== ENTERED CHILD PROCESS =====
Success for unrequested channel! [??]
==== MAIN PROCESS AFTER JOIN =====
Traceback (most recent call last):
  File "repro.py", line 22, in <module>
    stdin, stdout, stderr = client.exec_command("ls")
  File "/Users/vivejha/Projects/cisco/lib/python3.4/site-packages/paramiko/client.py", line 401, in exec_command
    chan = self._transport.open_session(timeout=timeout)
  File "/Users/vivejha/Projects/cisco/lib/python3.4/site-packages/paramiko/transport.py", line 702, in open_session
    timeout=timeout)
  File "/Users/vivejha/Projects/cisco/lib/python3.4/site-packages/paramiko/transport.py", line 823, in open_channel
    raise e
paramiko.ssh_exception.SSHException: Unable to open channel.

需要注意的一些重要事项

1. 如果我尝试在子进程中访问client。首先它根本不起作用。

2. 其次，主进程中的句柄也出人意料地死掉了。我不知道这种孩子与父母的交流是如何促进的以及为什么。

3. 最大的问题是程序最终挂起，异常很好，但最不希望挂起。

4. 如果我不在子进程中使用client，而做一些其他工作，那么父进程中的client 不会受到影响并且照常工作。

   李>

注意：transport.py 中有一个叫做atfork 的东西，它声称可以控制这种行为。但令人惊讶的是，即使在该方法中注释代码也没有影响。在 paramiko 的整个代码库中也没有对 atfork 的引用。

PS：我使用的是最新的 paramiko，这个程序是在 Mac 上运行的

Answer 1

fork 涉及到套接字时，这只是一个基本问题。两个进程共享同一个套接字，但只有一个可以使用它。试想一下，两个不同的进程正在管理一个套接字。他们都处于不同的状态，例如一个可能会向远程端发送和接收数据，而另一个则处于完全不同的加密状态。想想随机数/初始化向量，当两个进程分歧时，它们只会在分叉后无效。

你的问题的解决方案显然是从MultiProcessing切换到MultiThreading。这样，您只有一个跨所有线程共享的 ssh 连接。如果您真的想使用 fork，则必须在 fork 中为每个 fork 创建一个新连接。

见transport.py

def atfork(self):
    """
    Terminate this Transport without closing the session.  On posix
    systems, if a Transport is open during process forking, both parent
    and child will share the underlying socket, but only one process can
    use the connection (without corrupting the session).  Use this method
    to clean up a Transport object without disrupting the other process.

在 paramiko 日志中，您会看到您的父进程从远程端接收到 SSH_DISCONNECT_MSG，错误为：Packet corrupt。很可能是由于父节点处于不同的加密状态并发送服务器无法理解的数据包。

DEBUG:lala:==== ENTERED CHILD PROCESS =====
DEBUG:lala:<paramiko.SSHClient object at 0xb74bf1ac>
DEBUG:lala:<paramiko.Transport at 0xb6fed82cL (cipher aes128-ctr, 128 bits) (active; 0 open channel(s))>
DEBUG:paramiko.transport:[chan 1] Max packet in: 34816 bytes
WARNING:paramiko.transport:Success for unrequested channel! [??]


DEBUG:lala:==== MAIN PROCESS AFTER JOIN =====
WARNING:lala:<socket._socketobject object at 0xb706ef7c>
DEBUG:paramiko.transport:[chan 1] Max packet in: 34816 bytes
INFO:paramiko.transport:Disconnect (code 2): Packet corrupt

这是一个使用concurrent.futures 的基本多线程示例：

from concurrent.futures import ThreadPoolExecutor

def simple_work(handle):
    print("==== ENTERED CHILD PROCESS =====")
    stdin, stdout, stderr = handle.exec_command("whoami")
    print(stdout.read())
    print("==== EXITING CHILD PROCESS =====")

with ThreadPoolExecutor(max_workers=2) as executor:
    future = executor.submit(simple_work, client)
    print(future.result())

print("==== MAIN PROCESS AFTER JOIN =====")
stdin, stdout, stderr = client.exec_command("echo AFTER && whoami")
print(stdout.read())

还要注意，在大多数情况下，您甚至不需要引入额外的线程。 Paramiko exec_command 已经生成了一个新线程，并且在您尝试从任何伪文件 stdout、stderr 读取之前不会阻塞。这意味着，您也可以只执行一些命令并稍后从标准输出中读取。但请记住，paramiko 可能会由于缓冲区已满而停止。