用Python自动操作工具paramiko：通过SSH连接并远端执行服务器命令的第三方库指南
最编程
2024-02-10 10:18:22
...
前言
1、 paramiko 库是一个用于做远程控制的模块，使用该模块可以对远程服务器进行命令或文件操作，值得一说的是，fabric和ansible内部的远程管理就是使用的 paramiko 库来现实。（提供客户端和服务器的功能）
2、 paramiko 库是用python语言写的一个模块，遵循SSH2协议，支持以加密和认证的方式，进行远程服务器的连接。 paramiko 库支持Linux，Solaris，BSD，MacOS X，Windows等平台通过SSH从一个平台连接到另外一个平台。
3、利用 paramiko 模块，可以方便的进行ssh连接和sftp协议进行sftp文件传输。
4、 paramiko 模块包含了两个核心组件： SSHClient 和 SFTPClient 。
- ①SSHClient 类：SSHClient 类是与SSH服务器会话的高级表示。该类集成了Transport，Channel 和SFTPClient 类。（通过ssh协议和linux服务器建立连接执行命令，该类封装了传输(Transport)，通道(Channel)及SFTPClient建立的方法）
- ②SFTClient 类：该类通过一个打开的SSH Transport 会话创建SFTP会话通道并执行远程文件操作。（作用类似与Linux的sftp命令，是对SFTP客户端的封装，用以实现远程文件操作，如文件上传、下载、修改文件权限等操作。）
5、 paramiko 可以通过ssh协议执行远程主机的程序或脚本，获取输出结果和返回值，使用起来简洁优雅。
6、名词介绍：
Channel：是一种类Socket，一种安全的SSH传输通道； Transport：是一种加密的会话（但是这样一个对象的Session并未建立），并且创建了一个加密的tunnels，这个tunnels叫做Channel；
Channel：是一种类Socket，一种安全的SSH传输通道； Session：是client与Server保持连接的对象，用connect()/start_client()/start_server()开始会话； hostname(str类型)，连接的目标主机地址； port(int类型)，连接目标主机的端口，默认为22； username(str类型)，校验的用户名(默认为当前的本地用户名)； password(str类型)，密码用于身份校验或解锁私钥； pkey(Pkey类型)，私钥方式用于身份验证； key_filename(str or list(str)类型)，一个文件名或文件名列表，用于私钥的身份验证； timeout(float类型)，一个可选的超时时间(以秒为单位)的TCP连接； allow_agent(bool类型)，设置为False时用于禁用连接到SSH代理； look_for_keys(bool类型)，设置为False时用于来禁用在～/.ssh中搜索私钥文件； compress(bool类型)，设置为True时打开压缩。
下载安装
1、python3版本的下载与安装
pip3 install paramiko
2、python2版本的下载与安装
#在python2中，由于 paramiko 模块内部依赖pycrypto模块，所以先下载安装pycrypto pip3 install pycrypto pip3 install paramiko 注：如果在安装pycrypto2.0.1时发生如下错误 command 'gcc' failed with exit status 1...
可能是缺少python-dev安装包导致 如果gcc没有安装，请事先安装gcc
paramiko库中SSHClient模块的使用：
1、作用：用于连接远程服务器并执行基本命令。
2、远程连接分为两种：（1）基于用户名密码连接远程服务器 （2）基于公钥秘钥连接远程服务器。
3、通过使用 paramiko 库远程操作服务器，其实本质也分为两种：（1）使用 SSHClient 类 （2） SSHClient 类 封装 transport 类
基于用户名密码连接
连接远程服务器：
def connect( self, hostname, port=SSH_PORT, username=None, password=None, pkey=None, key_filename=None, timeout=None, allow_agent=True, look_for_keys=True, compress=False, sock=None, gss_auth=False, gss_kex=False, gss_deleg_creds=True, gss_host=None, banner_timeout=None, auth_timeout=None, gss_trust_dns=True, passphrase=None, disabled_algorithms=None, ): """ Connect to an SSH server and authenticate to it. The server's host key is checked against the system host keys (see `load_system_host_keys`) and any local host keys (`load_host_keys`). If the server's hostname is not found in either set of host keys, the missing host key policy is used (see `set_missing_host_key_policy`). The default policy is to reject the key and raise an `.SSHException`. Authentication is attempted in the following order of priority: - The ``pkey`` or ``key_filename`` passed in (if any) - ``key_filename`` may contain OpenSSH public certificate paths as well as regular private-key paths; when files ending in ``-cert.pub`` are found, they are assumed to match a private key, and both components will be loaded. (The private key itself does *not* need to be listed in ``key_filename`` for this to occur - *just* the certificate.) - Any key we can find through an SSH agent - Any "id_rsa", "id_dsa" or "id_ecdsa" key discoverable in ``~/.ssh/`` - When OpenSSH-style public certificates exist that match an existing such private key (so e.g. one has ``id_rsa`` and ``id_rsa-cert.pub``) the certificate will be loaded alongside the private key and used for authentication. - Plain username/password auth, if a password was given If a private key requires a password to unlock it, and a password is passed in, that password will be used to attempt to unlock the key. :param str hostname: the server to connect to :param int port: the server port to connect to :param str username: the username to authenticate as (defaults to the current local username) :param str password: Used for password authentication; is also used for private key decryption if ``passphrase`` is not given. :param str passphrase: Used for decrypting private keys. :param .PKey pkey: an optional private key to use for authentication :param str key_filename: the filename, or list of filenames, of optional private key(s) and/or certs to try for authentication :param float timeout: an optional timeout (in seconds) for the TCP connect :param bool allow_agent: set to False to disable connecting to the SSH agent :param bool look_for_keys: set to False to disable searching for discoverable private key files in ``~/.ssh/`` :param bool compress: set to True to turn on compression :param socket sock: an open socket or socket-like object (such as a `.Channel`) to use for communication to the target host :param bool gss_auth: ``True`` if you want to use GSS-API authentication :param bool gss_kex: Perform GSS-API Key Exchange and user authentication :param bool gss_deleg_creds: Delegate GSS-API client credentials or not :param str gss_host: The targets name in the kerberos database. default: hostname :param bool gss_trust_dns: Indicates whether or not the DNS is trusted to securely canonicalize the name of the host being connected to (default ``True``). :param float banner_timeout: an optional timeout (in seconds) to wait for the SSH banner to be presented. :param float auth_timeout: an optional timeout (in seconds) to wait for an authentication response. :param dict disabled_algorithms: an optional dict passed directly to `.Transport` and its keyword argument of the same name. :raises: `.BadHostKeyException` -- if the server's host key could not be verified :raises: `.AuthenticationException` -- if authentication failed :raises: `.SSHException` -- if there was any other error connecting or establishing an SSH session :raises socket.error: if a socket error occurred while connecting .. versionchanged:: 1.15 Added the ``banner_timeout``, ``gss_auth``, ``gss_kex``, ``gss_deleg_creds`` and ``gss_host`` arguments. .. versionchanged:: 2.3 Added the ``gss_trust_dns`` argument. .. versionchanged:: 2.4 Added the ``passphrase`` argument. .. versionchanged:: 2.6 Added the ``disabled_algorithms`` argument. """ if not sock: errors = {} # Try multiple possible address families (e.g. IPv4 vs IPv6) to_try = list(self._families_and_addresses(hostname, port)) for af, addr in to_try: try: sock = socket.socket(af, socket.SOCK_STREAM) if timeout is not None: try: sock.settimeout(timeout) except: pass retry_on_signal(lambda: sock.connect(addr)) # Break out of the loop on success break except socket.error as e: # Raise anything that isn't a straight up connection error # (such as a resolution error) if e.errno not in (ECONNREFUSED, EHOSTUNREACH): raise # Capture anything else so we know how the run looks once # iteration is complete. Retain info about which attempt # this was. errors[addr] = e # Make sure we explode usefully if no address family attempts # succeeded. We've no way of knowing which error is the "right" # one, so we construct a hybrid exception containing all the real # ones, of a subclass that client code should still be watching for # (socket.error) if len(errors) == len(to_try): raise NoValidConnectionsError(errors) t = self._transport = Transport( sock, gss_kex=gss_kex, gss_deleg_creds=gss_deleg_creds, disabled_algorithms=disabled_algorithms, ) t.use_compression(compress=compress) t.set_gss_host( # t.hostname may be None, but GSS-API requires a target name. # Therefore use hostname as fallback. gss_host=gss_host or hostname, trust_dns=gss_trust_dns, gssapi_requested=gss_auth or gss_kex, ) if self._log_channel is not None: t.set_log_channel(self._log_channel) if banner_timeout is not None: t.banner_timeout = banner_timeout if auth_timeout is not None: t.auth_timeout = auth_timeout if port == SSH_PORT: server_hostkey_name = hostname else: server_hostkey_name = "[{}]:{}".format(hostname, port) our_server_keys = None our_server_keys = self._system_host_keys.get(server_hostkey_name) if our_server_keys is None: our_server_keys = self._host_keys.get(server_hostkey_name) if our_server_keys is not None: keytype = our_server_keys.keys()[0] sec_opts = t.get_security_options() other_types = [x for x in sec_opts.key_types if x != keytype] sec_opts.key_types = [keytype] + other_types t.start_client(timeout=timeout) # If GSS-API Key Exchange is performed we are not required to check the # host key, because the host is authenticated via GSS-API / SSPI as # well as our client. if not self._transport.gss_kex_used: server_key = t.get_remote_server_key() if our_server_keys is None: # will raise exception if the key is rejected self._policy.missing_host_key( self, server_hostkey_name, server_key ) else: our_key = our_server_keys.get(server_key.get_name()) if our_key != server_key: if our_key is None: our_key = list(our_server_keys.values())[0] raise BadHostKeyException(hostname, server_key, our_key) if username is None: username = getpass.getuser() if key_filename is None: key_filenames = [] elif isinstance(key_filename, string_types): key_filenames = [key_filename] else: key_filenames = key_filename self._auth( username, password, pkey, key_filenames, allow_agent, look_for_keys, gss_auth, gss_kex, gss_deleg_creds, t.gss_host, passphrase, )
远程执行命令：
exec_command(command, bufsize=-1, timeout=None, get_pty=False, environment=None)
源码如下：
def exec_command( self, command, bufsize=-1, timeout=None, get_pty=False, environment=None, ): """ Execute a command on the SSH server. A new `.Channel` is opened and the requested command is executed. The command's input and output streams are returned as Python ``file``-like objects representing stdin, stdout, and stderr. :param str command: the command to execute :param int bufsize: interpreted the same way as by the built-in ``file()`` function in Python :param int timeout: set command's channel timeout. See `.Channel.settimeout` :param bool get_pty: Request a pseudo-terminal from the server (default ``False``). See `.Channel.get_pty` :param dict environment: a dict of shell environment variables, to be merged into the default environment that the remote command executes within. .. warning:: Servers may silently reject some environment variables; see the warning in `.Channel.set_environment_variable` for details. :return: the stdin, stdout, and stderr of the executing command, as a 3-tuple :raises: `.SSHException` -- if the server fails to execute the command .. versionchanged:: 1.10 Added the ``get_pty`` kwarg. """ chan = self._transport.open_session(timeout=timeout) if get_pty: chan.get_pty() chan.settimeout(timeout) if environment: chan.update_environment(environment) chan.exec_command(command) stdin = chan.makefile_stdin("wb", bufsize) stdout = chan.makefile("r", bufsize) stderr = chan.makefile_stderr("r", bufsize) return stdin, stdout, stderr
1、参数说明：
- command (str类型)，执行的命令串；
- bufsize (int类型)，文件缓冲区大小，默认为-1(不限制)
2、使用 exec_command() 方法执行命令会返回三个信息：
- 标准输入内容（用于实现交互式命令）--- stdin
- 标准输出（保存命令的正常执行结果）--- stdout
- 标准错误输出（保存命令的错误信息）--- stderr
3、通过 exec_command() 方法命令执行完毕后，通道将关闭，不能再使用。如果想继续执行另一个命令，必须打开一个新通道。
4、stdout.channel.recv_exit_status() 方法：用来判断命令是否执行完成，当 exec_command() 方法中的命令没有执行完成时，这个方法会一直阻塞。如果 exec_command() 方法运行结束，则会返回0表示执行成功。-1表示执行失败。
5、如果需要通过 exec_command() 方法连续的在远程服务器上执行命令，则command参数中多个命令需要使用 ;分号 隔开。（例如场景：需要进入远程服务器中的xx路径，随后在该路径中执行命令）。
ssh对象连接远程服务器并执行命令获取控制台打印的信息：
import paramiko # 创建SSH对象（ssh_clint） ssh_clint = paramiko.SSHClient() # 通过这个set_missing_host_key_policy方法用于实现远程登录是否需要确认输入yes，否则保存（相当于白名单的作用） ssh_clint.set_missing_host_key_policy(paramiko.AutoAddPolicy()) # 使用connect类来连接远程服务器 ssh_clint.connect(hostname='172.16.1.1', port=22, username='test', password='123') # 使用exec_command方法执行命令，并使用变量接收命令的返回值并用print输出 stdin, stdout, stderr = ssh_clint.exec_command('df') # 获取命令结果 result = stdout.read() print(result.decode('utf-8')) # 关闭连接 ssh_clint.close()
运行结果：
注意：
# 允许连接不在know_hosts文件中的主机 ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
如果在连接远程服务器之前缺少这部分代码，会报错：
报错的含义是：提示在服务器的known_hosts中没有，这个就是连接服务器的时候那个首次连接需要输入一个yes保存证书。（远程服务器没有本地主机密钥或HostKeys对象）
配置 set_missing_host_key_policy(policy) 方法的参数常见取值有三种：
AutoAddPolicy ：自动添加远程服务器的主机名及主机密钥到本地主机的known_hosts，不依赖load_system_host_key的配置。即新建立ssh连接时不需要再输入yes或no进行确认。最为常用。（将信任的主机自动加入到host_allow列表）
WarningPolicy ：用于记录一个未知的主机密钥的python警告。并接受，功能上和AutoAddPolicy类似，但是会提示是新连接。
RejectPolicy ： 自动拒绝未知的主机名和密钥，依赖load_system_host_key的配置。此为默认选项。
[解决本地主机首次连接连接远程服务器出现的known_hosts问题：通过 set_missing_host_key_policy(paramiko.AutoAddPolicy()) 方法用于实现ssh远程登录时是否需要确认输入yes，否则保存]
使用 recv_exit_status() 方法判断服务器上的命令是否已经执行完成
import paramiko # 建立一个sshclient对象 ssh = paramiko.SSHClient() # 将信任的主机自动加入到host_allow列表，须放在connect方法前面 ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) # 调用connect方法连接服务器 ssh.connect(hostname="172.16.1.166", port=22, username="test", password="123") # 执行命令 stdin, stdout, stderr = ssh.exec_command("echo `date` && df -hl") # 结果放到stdout中，如果有错误将放到stderr中 print(stdout.read().decode('utf-8')) # recv_exit_status方法会一直阻塞直到命令执行完成 return_code = stdout.channel.recv_exit_status() print("return_code:", return_code) # 关闭连接 ssh.close()
运行结果：
使用 exec_command() 方法连续的执行多个命令并获取终端输出
import paramiko # 创建SSH对象（ssh_clint） ssh_clint = paramiko.SSHClient() # 通过这个set_missing_host_key_policy方法用于实现登录是需要确认输入yes，否则保存 ssh_clint.set_missing_host_key_policy(paramiko.AutoAddPolicy()) # 使用connect类来连接远程服务器 try: ssh_clint.connect(hostname='172.16.0.190', port=22, username='test', password='123') except Exception as e: raise # 使用exec_command方法执行命令，并使用变量接收命令的返回值并用print输出 stdin, stdout, stderr = ssh_clint.exec_command( command='pwd; cd /home/test/PycharmProjects/hls_practice/practice_python; pwd; ls -lh') # 获取命令结果 result = stdout.read() status = stdout.channel.recv_exit_status() print('1、命令是否成功执行完成的结果：（0表示执行成功，非0表示执行失败）', status) print('2、远程服务器上成功执行命令后的控制台输出结果：\n', result.decode('utf-8')) # 关闭连接 ssh_clint.close()
运行结果：
使用try-except捕获异常：
import paramiko import sys # 定义函数ssh,把操作内容写到函数里 def sshExeCMD(): # 定义一个变量ssh_clint使用SSHClient类用来后边调用 ssh_client = paramiko.SSHClient() # 通过这个set_missing_host_key_policy方法用于实现登录是需要确认输入yes，否则保存 ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) # 使用try做异常捕获 try: # 使用connect类来连接服务器 ssh_client.connect(hostname="192.168.1.110", port=22, username="test", password="123") # 如果上边命令报错吧报错信息定义到err变量，并输出。 except Exception as err: print("服务器链接失败！！！") print(err) # 如果报错使用sys的exit退出脚本 sys.exit() # 使用exec_command方法执行命令，并使用变量接收命令的返回值并用print输出 stdin, stdout, stderr = ssh_client.exec_command("df -hT") print(str(stdout.read())) # 通过判断模块名运行上边函数 if __name__ == '__main__': sshExeCMD()
运行结果：（错误的远程服务器主机地址）
运行结果2：（错误的远程服务器主机登录密码）
使用paramiko库一次性执行多个命令有以下两种方式：
1、使用SSHClient对象的 invoke_shell 方法创建一个交互式的终端，然后向终端依次发送多条完整的命令，最后关闭连接即可。
示例代码如下：
import paramiko # 创建SSHClient对象 ssh = paramiko.SSHClient() # 密钥认证 ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) ssh.connect(hostname='服务器IP地址', port=22, username='用户名', key_filename='私钥文件路径') # 创建交互式终端 channel = ssh.invoke_shell() # 发送多个命令 channel.send("command1\n") channel.send("command2\n") channel.send("command3\n") # 关闭终端和连接 channel.close() ssh.close()
2、在一个SSH会话中，使用exec_command方法执行一条长命令，其中多个子命令之间使用分号 ';' 或者并列排列。
示例代码如下：
import paramiko # 创建SSHClient对象 ssh = paramiko.SSHClient() # 密钥认证 ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) ssh.connect(hostname='服务器IP地址', port=22, username='用户名', key_filename='私钥文件路径') # 执行长命令 stdin, stdout, stderr = ssh.exec_command('command1; command2; command3') print(stdout.read().decode()) # 关闭连接 ssh.close()
多台远程服务器上执行命令：
# 导入paramiko，（导入前需要先在环境里安装该模块） import paramiko import sys # 定义函数ssh,把操作内容写到函数里,函数里接收参数（写在括号里）,其中port=是设置一个默认值如果没传就用默认 def sshExeCMD(ip, username, password, port=22): # 定义一个变量ssh_clint使用SSHClient类用来后边调用 ssh_client = paramiko.SSHClient() # 通过这个set_missing_host_key_policy方法用于实现登录是需要确认输入yes，否则保存 ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) # 使用try做异常捕获 try: # 使用connect类来连接服务器 ssh_client.connect(hostname=ip, port=port, username=username, password=password) # 如果上边命令报错吧报错信息定义到err变量，并输出。 except Exception as err: print("服务器链接失败！！！" % ip) print(err) # 如果报错使用sys的exit退出脚本 sys.exit() # 使用exec_command方法执行命令，并使用变量接收命令的返回值并用print输出 # 这里也可以把命令做成参数传进来 stdin, stdout, stderr = ssh_client.exec_command("hostname") # 使用decode方法可以指定编码 print(stdout.read().decode("utf-8")) # 通过判断模块名运行上边函数 if __name__ == '__main__': # 定义一个字典，写服务器信息 servers = { # 以服务器IP为键,值为服务器的用户密码端口定义的字典 "192.168.1.110": { "username": "songxk", "password": "123123", "port": 22, }, "192.168.1.123": { "username": "root", "password": "123123", "port": 22, }, } # 使用items方法遍历，使用ip 和info把字典里的键和值遍历赋值，传给函数sshExeCMD for ip, info in servers.items(): # 这里的info也就是上边定义的ip对应值的字典，使用get获取这个字典里对应username键对应的值，赋值给变量username传给函数中使用 sshExeCMD( ip=ip, username=info.get("username"), password=info.get("password"), port=info.get("port") )
SSHClient 封装 Transport 连接远程服务器：
import paramiko # 获取transport对象，配置主机名，端口 transport = paramiko.Transport(('172.16.1.1', 22))
# 设置登录名、密码 transport.connect(username='test', password='123')
# 获取ssh_client对象 ssh_client = paramiko.SSHClient() ssh_client._transport = transport # 获取远程服务器的主机名 stdin, stdout, stderr = ssh_client.exec_command('hostname') res = stdout.read() print(res.decode('utf-8')) transport.close()
运行结果：
基于公钥密钥连接远程服务器（本地主机通过私钥加密加密，远程服务器通过公钥解密）
1、客户端文件名：id_rsa（id_rsa文件是本地主机的私钥，使用密钥连接远程服务器时，必须要在远程服务器上配制公钥）
2、服务端必须有文件名：authorized_keys（authorized_keys文件是远程服务器上的公钥）(在用ssh-keygen时，必须制作一个authorized_keys，可以用ssh-copy-id来制作)
3、id_rsa文件的来源：终端执行命令 ssh-keygen ，然后一直回车，最后会在~/.ssh目录下生成该文件。
ssh客户端通过使用密钥连接远程服务器并执行命令获取控制台打印信息：
import paramiko private_key = paramiko.RSAKey.from_private_key_file('/tmp/id_rsa') # 创建SSH对象 ssh = paramiko.SSHClient() # 允许连接不在know_hosts文件中的主机 ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) # 连接服务器 ssh.connect(hostname='120.92.84.249', port=22, username='root', pkey=private_key) # 执行命令 stdin, stdout, stderr = ssh.exec_command('df') # 获取命令结果 result = stdout.read() print(result.decode('utf-8')) # 关闭连接 ssh.close()
SSHClient 封装 Transport 连接远程服务器：
import paramiko private_key = paramiko.RSAKey.from_private_key_file('/tmp/id_rsa') # 获取transport对象，配置主机名，端口 transport = paramiko.Transport(('120.92.84.249', 22)) # 设置登录名、连接远程服务器的登录密钥 transport.connect(username='root', pkey=private_key) # 获取ssh_client对象 ssh = paramiko.SSHClient() ssh._transport = transport # 获取远程服务器的主机名 stdin, stdout, stderr = ssh.exec_command('hostname') result = stdout.read() print(result.decode('utf-8')) # 关闭连接 transport.close()