使用Paramiko在Python中实现基于ssh的交互式shell？

Question

使用Paramiko在Python中实现基于ssh的交互式shell？

22

我想编写一个程序（在Windows 7上使用Python 3.x），通过ssh在远程shell上执行多个命令。经过查看paramiko的exec_command()函数，我意识到它不适用于我的用例（因为通道在命令执行后关闭），由于命令依赖于环境变量（由之前的命令设置），不能将它们连接成一个exec_command()调用，因为它们要在程序中的不同时间执行。

因此，我想在同一个通道中执行命令。接下来我尝试了使用paramiko的invoke_shell()函数实现交互式shell：

ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect(host, username=user, password=psw, port=22)

channel = ssh.invoke_shell()

out = channel.recv(9999)

channel.send('cd mivne_final\n')
channel.send('ls\n')

while not channel.recv_ready():
    time.sleep(3)

out = channel.recv(9999)
print(out.decode("ascii"))

channel.send('cd ..\n')
channel.send('cd or_fail\n')
channel.send('ls\n')

while not channel.recv_ready():
    time.sleep(3)

out = channel.recv(9999)
print(out.decode("ascii"))

channel.send('cd ..\n')
channel.send('cd simulator\n')
channel.send('ls\n')

while not channel.recv_ready():
    time.sleep(3)

out = channel.recv(9999)
print(out.decode("ascii"))

ssh.close()

这个代码存在一些问题：

第一个 print 并不总是打印 ls 的输出（有时只在第二个 print 中才打印出来）。
虽然我通过 recv 命令获取输出，但第一个 cd 和 ls 命令总是出现在输出中，而所有后续的 cd 和 ls 命令有时会被打印出来，有时则不会。
当被打印出来时，第二个和第三个 cd 和 ls 命令总是出现在第一个 ls 输出之前。

我对这种“非确定性”感到困惑，请您给予帮助，谢谢。

- misha

1

如果您将最少关注者的标签替换为Python标签（假设这确实是Python代码），那么您将获得更多帮助。祝你好运。 - shellter

你必须使用 paramiko 吗？我发现使用 fabric 更容易。你只需要设置 env 变量，如 user、password 和 host_string，然后就可以做各种事情，比如使用 get 从远程主机下载文件，使用 put 发送文件，以及使用 run 发出命令。你可以像这样链接命令：run('cd .. && cd simulator && ls')。 - kchomski

很遗憾，Fabric与Python 3.x不兼容，因此它不是一个选项。无论如何，从我所看到的来看，Fabric只是paramiko的包装器，并且不允许我在同一通道中运行“非链接”命令。最终我想要在shell命令之间运行很多逻辑。 - misha

@misha：抱歉，我忽略了你正在使用Python 3.x。 - kchomski

请查看netmiko。它专门用于网络设备，但您也可以在Linux上使用它。它适用于Python 3，并建立在Paramiko之上，但会为您处理大量缓冲。 - Ben

请注意，使用invoke_shell意味着远程服务器可能会发送改变用户终端字体颜色等字符序列，清除这些混乱是一件麻烦的事情。 - chrisinmtown

3个回答

1

我需要这个用于Cisco路由器，它与Linux机器有很大的区别。感谢@misha在这里识别出的一个重大挑战，即检测命令输出的结束，以便知道何时停止从路由器的stdout对象读取。如果你没有检测到这一点，读取循环就会一直挂起。每次发送两个命令，并将第二个命令用作哨兵是一种聪明的技巧，所以我也借鉴了这个方法！这使用了IOS命令提示符的一个众所周知的错误响应作为哨兵。

import logging
import paramiko
import socket

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)

# not provided here: inter_handler
# a custom authentication handler with proprietary deteails


def ssh_run_cmds(
        host: str,
        port: int,
        user: str,
        commands: list) -> None:
    """
    Connect to the router, authenticate by computing a challenge
    response, and run commands. A major challenge is detecting
    the end of the command output, to know when to stop reading
    from the router session. This code uses an ugly hack of
    sending an invalid command and checking for a well-known
    error message.
    """
    # Create a socket and connect it to port 22 on the remote host
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    # the argument must be a tuple
    sock.connect((host, port))
    # Wrap the socket in a paramiko Transport object
    ts = paramiko.Transport(sock)
    # Tell Paramiko that the Transport is going to be used as a client
    ts.start_client(timeout=10)
    # Authenticate the specified user via the handler
    ts.auth_interactive(user, inter_handler)
    # Open a channel
    chan = ts.open_channel(kind='session', timeout=10)
    # Associate a pseudo tty
    chan.get_pty()
    # Request an interactive shell session
    chan.invoke_shell()
    # Create writer/reader file-like objects
    stdin = chan.makefile('wb')
    stdout = chan.makefile('r')
    # Use the output from this invalid command as a sentinel
    bogus_cmd = 'show bogus'
    for cmd in commands:
        # Send the command AND a bogus command to detect end of output
        cmds = f'{cmd}\n{bogus_cmd}\n'
        logger.debug('Send commands: %s', cmds)
        stdin.write(cmds)
        stdin.flush()

        # Read the response
        for line in stdout:
            line = line.strip()
            logger.debug('Output line: %s', line)
            # the response from the bogus command is the last line
            if line.startswith("% Invalid input detected at '^' marker."):
                break
        # for line
    # for cmd

    stdin.close()
    stdout.close()
    chan.close()
    ts.close()
    sock.close()

- chrisinmtown

0

我尝试了上面的答案，但它不起作用，因为在我使用SSH与Python CLI时，ECHO命令返回错误。

因此，我编写了另一段代码，适用于Python CLI，假设输出是单行的。

我还认为类似于f"print('{finish}')"的东西可以像上面的答案中的ECHO一样完成同样的事情（终止符？？）。但是我没有利用它，因为我的输出始终必须是单行的。

class MusicPlayer:
def __init__(self, host='', username='pi', password=''):
    self.ssh = paramiko.SSHClient()
    self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    self.ssh.connect(host, username=username, password=password)
    channel = self.ssh.invoke_shell()
    self.stdin = channel.makefile('wb')
    self.stdout = channel.makefile('r')
    self.in_history = []
    self.out_history = []
    self.init_vlc()
    self.print()
    # atexit.register(self.__del__)

def __del__(self):
    self.ssh.close()

def execute(self, cmd):
    self.in_history.append(cmd)
    self.stdin.write(cmd + '\n')

def print(self, lines=1):
    for line in self.stdout:
        lined = line.strip()
        print(lined)
        self.out_history.append(lined)
        if self.in_history[-1] in lined:
            next_one = self.stdout.__next__().strip()
            print(next_one)
            self.out_history.append(next_one)
            return next_one

def init_vlc(self):
    for command in ['python', 'import vlc', 'import time', 'media_player = vlc.MediaPlayer()']:
        self.execute(command)

- Junmin Lee

网页内容由stack overflow 提供, 点击上面的

可以查看英文原文，
原文链接

- misha · Accepted Answer

import paramiko
import re


class ShellHandler:

    def __init__(self, host, user, psw):
        self.ssh = paramiko.SSHClient()
        self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        self.ssh.connect(host, username=user, password=psw, port=22)

        channel = self.ssh.invoke_shell()
        self.stdin = channel.makefile('wb')
        self.stdout = channel.makefile('r')

    def __del__(self):
        self.ssh.close()

    def execute(self, cmd):
        """

        :param cmd: the command to be executed on the remote computer
        :examples:  execute('ls')
                    execute('finger')
                    execute('cd folder_name')
        """
        cmd = cmd.strip('\n')
        self.stdin.write(cmd + '\n')
        finish = 'end of stdOUT buffer. finished with exit status'
        echo_cmd = 'echo {} $?'.format(finish)
        self.stdin.write(echo_cmd + '\n')
        shin = self.stdin
        self.stdin.flush()

        shout = []
        sherr = []
        exit_status = 0
        for line in self.stdout:
            if str(line).startswith(cmd) or str(line).startswith(echo_cmd):
                # up for now filled with shell junk from stdin
                shout = []
            elif str(line).startswith(finish):
                # our finish command ends with the exit status
                exit_status = int(str(line).rsplit(maxsplit=1)[1])
                if exit_status:
                    # stderr is combined with stdout.
                    # thus, swap sherr with shout in a case of failure.
                    sherr = shout
                    shout = []
                break
            else:
                # get rid of 'coloring and formatting' special characters
                shout.append(re.compile(r'(\x9B|\x1B\[)[0-?]*[ -/]*[@-~]').sub('', line).
                             replace('\b', '').replace('\r', ''))

        # first and last lines of shout/sherr contain a prompt
        if shout and echo_cmd in shout[-1]:
            shout.pop()
        if shout and cmd in shout[0]:
            shout.pop(0)
        if sherr and echo_cmd in sherr[-1]:
            sherr.pop()
        if sherr and cmd in sherr[0]:
            sherr.pop(0)

        return shin, shout, sherr