Интеграционное тестирование Python: использование paramiko для подделки ssh-сервера и внесения в журнал заданной ему команды

Question

Вот контекст: я пишу процедуру, которая подключается к удаленному серверу через SSH (с paramiko), выполняет некоторые команды и затем возвращает результат команд в процедуру.Он работает так же хорошо, как я могу сказать.

Я сейчас пытаюсь сделать тест "конец в конец", подделав сервер ssh, чтобы проверить, отправлены ли команды ssh, и чтоих результаты возвращаются к процедуре.

Я нашел поддельный SSH-сервер, который использует paramiko, он хорошо работает, когда я подключаюсь к нему через настоящий SSH-клиент, но у меня возникают ошибки при попытке подключиться кэто с paramiko ssh клиентом: /

Вы знаете, что я делаю неправильно?

Вот мой код:

Сервер:

#!/usr/bin/env python
"""Fake SSH Server Utilizing Paramiko"""
import threading
import socket
import sys
import traceback
import paramiko
import os

__location__ = os.path.realpath(os.path.join(os.getcwd(), os.path.dirname(__file__)))
LOG = open("log.txt", "a")
#HOST_KEY = paramiko.RSAKey(filename='keys/private.key')
HOST_KEY = paramiko.RSAKey(filename=os.path.join(__location__, 'id_rsa.new'))
PORT = 2200


def handle_cmd(cmd, chan):
    """Branching statements to handle and prepare a response for a command"""
    response = ""
    if cmd.startswith("sudo"):
        send_ascii("sudo.txt", chan)
        return
    elif cmd.startswith("ls"):
        response = "pw.txt"
    elif cmd.startswith("version"):
        response = "Super Amazing Awesome (tm) Shell v1.1"
    elif cmd.startswith("pwd"):
        response = "/home/clippy"
    elif cmd.startswith("cd"):
        send_ascii("cd.txt", chan)
        return
    elif cmd.startswith("cat"):
        send_ascii("cat.txt", chan)
        return
    elif cmd.startswith("rm"):
        send_ascii("bomb.txt", chan)
        response = "You blew up our files! How could you???"
    elif cmd.startswith("whoami"):
        send_ascii("wizard.txt", chan)
        response = "You are a wizard of the internet!"
    elif ".exe" in cmd:
        response = "Hmm, trying to access .exe files from an ssh terminal..... Your methods are unconventional"
    elif cmd.startswith("cmd"):
        response = "Command Prompt? We only use respectable shells on this machine.... Sorry"
    elif cmd == "help":
        send_ascii("help.txt", chan)
        return
    else:
        send_ascii("clippy.txt", chan)
        response = "Use the 'help' command to view available commands"

    LOG.write(response + "\n")
    LOG.flush()
    chan.send(response + "\r\n")


def send_ascii(filename, chan):
    """Print ascii from a file and send it to the channel"""
    with open('ascii/{}'.format(filename)) as text:
        chan.send("\r")
        for line in enumerate(text):
            LOG.write(line[1])
            chan.send(line[1] + "\r")
    LOG.flush()


class FakeSshServer(paramiko.ServerInterface):
    """Settings for paramiko server interface"""
    def __init__(self):
        self.event = threading.Event()

    def check_channel_request(self, kind, chanid):
        if kind == 'session':
            return paramiko.OPEN_SUCCEEDED
        return paramiko.OPEN_FAILED_ADMINISTRATIVELY_PROHIBITED

    def check_auth_password(self, username, password):
        # Accept all passwords as valid by default
        return paramiko.AUTH_SUCCESSFUL

    def get_allowed_auths(self, username):
        return 'password'

    def check_channel_shell_request(self, channel):
        self.event.set()
        return True

    def check_channel_pty_request(self, channel, term, width, height, pixelwidth, pixelheight, modes):
        return True


def start_server():
    """Init and run the ssh server"""
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        sock.bind(('', PORT))
    except Exception as err:
        print('*** Bind failed: {}'.format(err))
        traceback.print_exc()
        sys.exit(1)

    while True:
        try:
            sock.listen(100)
            print('Listening for connection ...')
            client, addr = sock.accept()
        except Exception as err:
            print('*** Listen/accept failed: {}'.format(err))
            traceback.print_exc()

        LOG.write("\n\nConnection from: " + addr[0] + "\n")
        print('Got a connection!')
        try:
            transport = paramiko.Transport(client)
            transport.add_server_key(HOST_KEY)
            # Change banner to appear legit on nmap (or other network) scans
            transport.local_version = "SSH-2.0-OpenSSH_7.6p1 Ubuntu-4ubuntu0.3"
            server = FakeSshServer()
            try:
                transport.start_server(server=server)
            except paramiko.SSHException:
                print('*** SSH negotiation failed.')
                raise Exception("SSH negotiation failed")
            # wait for auth
            chan = transport.accept(20)
            if chan is None:
                print('*** No channel.')
                raise Exception("No channel")

            server.event.wait(10)
            if not server.event.is_set():
                print('*** Client never asked for a shell.')
                raise Exception("No shell request")

            try:
                chan.send("Welcome to the my control server\r\n\r\n")
                run = True
                while run:
                    chan.send("$ ")
                    command = ""
                    while not command.endswith("\r"):
                        transport = chan.recv(1024)
                        # Echo input to psuedo-simulate a basic terminal
                        chan.send(transport)
                        command += transport.decode("utf-8")

                    chan.send("\r\n")
                    command = command.rstrip()
                    LOG.write("$ " + command + "\n")
                    print(command)
                    if command == "exit":
                        run = False
                    else:
                        handle_cmd(command, chan)

            except Exception as err:
                print('!!! Exception: {}: {}'.format(err.__class__, err))
                traceback.print_exc()
                try:
                    transport.close()
                except Exception:
                    pass

            chan.close()

        except Exception as err:
            print('!!! Exception: {}: {}'.format(err.__class__, err))
            traceback.print_exc()
            try:
                transport.close()
            except Exception:
                pass


if __name__ == "__main__":
    start_server()

Вот код, который я пытался подключить к нему с помощью paramiko:

    sshClient = paramiko.SSHClient()
    sshClient.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    sshClient.connect('127.0.0.1', username='bnc', password='pass', port=2200)

    sshClient.exec_command('ls')

, а также

    sshClient = paramiko.SSHClient()
    sshClient.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    sshClient.connect('127.0.0.1', username='bnc', password='pass', port=2200)

    channel = sshClient.get_transport().open_session()
    channel.get_pty()
    channel.invoke_shell()
    channel.send('ls\r\n')

с обоими тестами, я получил ту же ошибку:

Listening for connection ...
Got a connection!
!!! Exception: <class 'socket.error'>: Socket is closed
Traceback (most recent call last):
Listening for connection ...
  File "/home/cbrunain/Projects/daa/Python/test/ssh_mock_server.py", line 152, in start_server
    chan.send(transport)
  File "/home/cbrunain/.local/share/virtualenvs/daa-hnBs4Nn2/lib/python2.7/site-packages/paramiko/channel.py", line 801, in send
    return self._send(s, m)
  File "/home/cbrunain/.local/share/virtualenvs/daa-hnBs4Nn2/lib/python2.7/site-packages/paramiko/channel.py", line 1198, in _send
    raise socket.error("Socket is closed")
error: Socket is closed
No handlers could be found for logger "paramiko.transport"