Paramiko Sessions закрывает транспорт в дочернем процессе

Это просто фундаментальная проблема, когда сокеты связаны с fork. Оба процесса используют один и тот же сокет, но использовать его может только один. Просто представьте, что два разных процесса управляют одним сокетом. Они оба находятся в разных состояниях, например. один может отправлять и получать данные на удаленную сторону, в то время как другой находится в совершенно другом криптосостоянии. Просто подумайте о векторах одноразовых номеров/инициализации, они просто будут недействительными после разветвления, когда оба процесса расходятся.

Решение вашей проблемы, очевидно, заключается в переключении с MultiProcessing на MultiThreading. Таким образом, у вас есть только одно ssh-соединение, которое используется всеми потоками. Если вы действительно хотите использовать форк, вам придется разветвляться, создавая одно новое соединение для каждого форка.

см. 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 может зависнуть из-за переполнения буферов.