运河经济开发区建设局网站/站长工具的网址

问题

我们希望在windows或者linux上，可以使用ssh连接远程服务器，并且能够执行一般的linux命令，同时还要能够有一定交互能力。比如需要切换root用户，输入管理员用户密码等。

解决方案

Python的paramiko库，可以支持。但实现也有挺多问题需要考虑。主要有以下几点内容： + 命令执行，能够获取命令结果 + 命令执行，能够支持指定的预期结果 + 命令执行，要有超时能力，不能挂死。

用法1：

ssh = Ssh2Client('127.0.0.1', 22)

ssh.connect('root', 'xxxx')

result = ssh.exec('pwd')

print(result)

用法2：

ssh = Ssh2Client('127.0.0.1', 22)

ssh.connect('user-name', 'user-pwd')

ssh.exec('sudo su -', 'Password:')

ssh.exec('root-pwd')

ssh.exec('ls -l /var/root')

代码实现如下所示：

import re

import socket

import time

import paramiko

class Ssh2Client:

def __init__(self, host: str, port: int):

self.__host = host

self.__port = port

self.__ssh = None

self.__channel = None

def __del__(self):

self.__close()

def connect(self, user: str, pwd: str) -> bool:

self.__close()

self.__ssh = paramiko.SSHClient()

self.__ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

self.__ssh.connect(self.__host, username=user, password=pwd, port=self.__port)

return True

def exec(self, cmd: str, end_str=('# ', '$ ', '? ', '% '), timeout=30) -> str:

if not self.__channel:

self.__channel = self.__ssh.invoke_shell()

time.sleep(0.020)

self.__channel.recv(4096).decode()

if cmd.endswith('\n'):

self.__channel.send(cmd)

else:

self.__channel.send(cmd + '\n')

return self.__recv(self.__channel, end_str, timeout)

def __recv(self, channel, end_str, timeout) -> str:

result = ''

out_str = ''

max_wait_time = timeout * 1000

channel.settimeout(0.05)

while max_wait_time > 0:

try:

out = channel.recv(1024 * 1024).decode()

if not out or out == '':

continue

out_str = out_str + out

match, result = self.__match(out_str, end_str)

if match is True:

return result.strip()

else:

max_wait_time -= 50

except socket.timeout:

max_wait_time -= 50

raise Exception('recv data timeout')

def __match(self, out_str: str, end_str: list) -> (bool, str):

result = out_str

for it in end_str:

if result.endswith(it):

return True, result

return False, result

def __close(self):

if not self.__ssh:

return

self.__ssh.close()

self.__ssh = None

讨论

我们使用用法1，输出类似如下格式(用户名做了处理)：

pwd

/Users/user1

[xxx:~] xxx%

这里有两个问题要处理，命令和命令提示符都一并输出了。我们需要做特殊处理。处理方法也很简单，第一行和最后一行直接去掉即可，同时考虑命令无结果输出的处理即可。修改exec方法如下：

def exec(self, cmd: str, end_str=('# ', '$ ', '? ', '% '), timeout=30) -> str:

# ...

# 以下是新增的代码

result = self.__recv(self.__channel, end_str, timeout)

begin_pos = result.find('\r\n')

end_pos = result.rfind('\r\n')

if begin_pos == end_pos:

return ''

return result[begin_pos + 2:end_pos]

现状输出结果就正确了，这个就是我们想要的结果。

/Users/user1

偶然的机会，测试输入的命令比较长，取得结果又不正确了。比如执行

ssh.exec('echo 0111111111111111111111111111111111112222222222222222222222222222233333333333333333333333444444444444444')

输出结果，有的服务器，会返回下面这个奇怪的结果：

2222222222233333333333333333333333444444444444444

ls: 0111111111111111111111111111111111112222222222222222222222222222233333333333333333333333444444444444444: No such file or directory

这个问题的原因，主要是因为ssh2输出时使用了窗口的概念，默认是80*24，输入命令如果超过长度，会自动换行，导致处理命令结果时出错，主要修改invoke_shell函数调用方式，代码如下：

def exec(self, cmd: str, end_str=('# ', '$ ', '? ', '% '), timeout=30) -> str:

if not self.__channel:

# width和height，可以指定输出窗口的大小。

self.__channel = self.__ssh.invoke_shell(term='xterm', width=4096)

time.sleep(0.020)

self.__channel.recv(4096).decode()

# ....

命令窗口的宽度设置为4096，输出结果就对了。不过如果命令超过4096，输出还会出问题，根据实际情况，设置width的值，可以设置更大一点。

ls: 0111111111111111111111111111111111112222222222222222222222222222233333333333333333333333444444444444444: No such file or directory

到目前为止，已经基本够用了。但是还有一个问题，试用ls命令返回的结果，有一些奇怪的转义字符，比如：

[1;34mCacheVolume[0m [1;34mbin[0m [1;34mboot[0m [1;34mdev[0m [1;34metc[0m [1;34mhome[0m [1;34mlib[0m [1;36mlinuxrc[0m [1;34mlost+found[0m [1;34mmnt[0m [1;36mnfs[0m [1;34mopt[0m [1;

这个问题的处理比较麻烦，处理了很久也不行。开始使用字符串分析处理，忽略这些转义符，但总是有点麻烦，处理不够彻底。后来终于在网上搜索到，这个转义是叫ansi转义码，可以在term上显示彩色。网上给出了正则处理方法：

# 7-bit C1 ANSI sequences

self.__ansi_escape = re.compile(r'''\x1B # ESC(?: # 7-bit C1 Fe (except CSI)[@-Z\\-_]| # or [ for CSI, followed by a control sequence\[[0-?]* # Parameter bytes[ -/]* # Intermediate bytes[@-~] # Final byte)''', re.VERBOSE)

def __match(self, out_str: str, end_str: list) -> (bool, str):

result = self.__ansi_escape.sub('', out_str)

for it in end_str:

if result.endswith(it):

return True, result

return False, result

正则表达式比较复杂，有兴趣的同学自己分析这个re。

到目前为止，Ssh2Client已经基本实现，而且比较实用。可以处理绝大多数问题，实现也不复杂，比网上很多帖子都讲得全一些，代码可以直接拿来用。

但也不并是全部问题都能解决。比如有的linux系统，命令输出会出现换行，中文处理，都容易会导致输出结果获取不正确。不过，这些基本就是字符串分析和解码问题了。

完整的代码如下：

import re

import socket

import time

import paramiko

class Ssh2Client:

def __init__(self, host: str, port: int):

self.__host = host

self.__port = port

self.__ssh = None

self.__channel = None

# 7-bit C1 ANSI sequences

self.__ansi_escape = re.compile(r'''\x1B # ESC(?: # 7-bit C1 Fe (except CSI)[@-Z\\-_]| # or [ for CSI, followed by a control sequence\[[0-?]* # Parameter bytes[ -/]* # Intermediate bytes[@-~] # Final byte)''', re.VERBOSE)

def __del__(self):

self.__close()

def connect(self, user: str, pwd: str) -> bool:

self.__close()

self.__ssh = paramiko.SSHClient()

self.__ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

self.__ssh.connect(self.__host, username=user, password=pwd, port=self.__port)

return True

def exec(self, cmd: str, end_str=('# ', '$ ', '? ', '% '), timeout=30) -> str:

if not self.__channel:

self.__channel = self.__ssh.invoke_shell(term='xterm', width=4096, height=48)

time.sleep(0.020)

self.__channel.recv(4096).decode()

if cmd.endswith('\n'):

self.__channel.send(cmd)

else:

self.__channel.send(cmd + '\n')

result = self.__recv(self.__channel, end_str, timeout)

begin_pos = result.find('\r\n')

end_pos = result.rfind('\r\n')

if begin_pos == end_pos:

return ''

return result[begin_pos + 2:end_pos]

def __recv(self, channel, end_str, timeout) -> str:

result = ''

out_str = ''

max_wait_time = timeout * 1000

channel.settimeout(0.05)

while max_wait_time > 0:

try:

out = channel.recv(1024 * 1024).decode()

if not out or out == '':

continue

out_str = out_str + out

match, result = self.__match(out_str, end_str)

if match is True:

return result.strip()

else:

max_wait_time -= 50

except socket.timeout:

max_wait_time -= 50

raise Exception('recv data timeout')

def __match(self, out_str: str, end_str: list) -> (bool, str):

result = self.__ansi_escape.sub('', out_str)

for it in end_str:

if result.endswith(it):

return True, result

return False, result

def __close(self):

if not self.__ssh:

return

self.__ssh.close()

self.__ssh = None