我想编写一个程序（在 Windows 7 上使用 Python 3.x），通过 ssh 在远程 shell 上执行多个命令。在查看了 paramikos 的
exec_command()exec_command()因此，我想在同一个通道中执行命令。我研究的下一个选项是使用 paramikos 的
invoke_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()
这段代码有一些问题：
printlsprintcdlsrecvcdlscdlsls我对这种“非决定论”感到困惑，非常感谢您的帮助。
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
我需要这个用于 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()
我尝试了上面的答案，但它不起作用，因为 ECHO 命令在 Python CLI 中返回错误，而我将其与 SSH 一起使用。
所以我编写了另一段适用于 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)