python中连接linux好用的模块paramiko（附带案例）

Posted 2022-06-23 cui_yonghua

一. paramiko简介

paramiko是用python语言写的一个模块，遵循SSH2协议，支持以加密和认证的方式，进行远程服务器的连接。paramiko支持Linux, Solaris, BSD, MacOS X, Windows等平台通过SSH从一个平台连接到另外一个平台。利用该模块，可以方便的进行ssh连接和sftp协议进行sftp文件传输。

在python项目中需要远程操作linux服务器，在服务器上进行文件的拷贝以及linux命令操作。给python安装paramiko 远程连接操作linux服务器是很好的选择。

安装：pip3 install paramiko
paramiko库有两种连接主机的方式，一种是使用用户名和密码，一种是使用秘钥连接。
当使用用户名密码连接时有个坑需要注意：
当你连接的对方主机有SSH秘钥的时候，在使用用户名密码连接的时候

ValueError: ('Invalid private key', [_OpenSSLErrorWithText(code=67764350, lib=4, func=160, reason=126, reason_text=b'error:040A007E:rsa routines:RSA_check_key_ex:iqmp not inverse of q')])

通常更新版本即可

二. 3个使用案例

2.1 连接方式一：

这种方式比较简单，没有进行封装。单纯的实现了远程连接linux服务器。下面还有一种方式，是在这个方式的基础上将每个主要步骤做了封装，方便业务调用。

import paramiko
 
def remote_connect():
    # 服务器相关信息,下面输入你个人的用户名、密码、ip等信息
    ip = "192.168.1.110"
    port = 22
    user = "root"
    password = "123456"

    # 创建SSHClient 实例对象
    ssh = paramiko.SSHClient()
    # 调用方法，表示没有存储远程机器的公钥，允许访问
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    # 连接远程机器，地址，端口，用户名密码
    ssh.connect(ip, port, user, password, timeout=10)
    # 输入linux命令
    ls = "ls"
    stdin, stdout, stderr = ssh.exec_command(ls)
    # 输出命令执行结果
    result = stdout.read()
    print(result)
    # 关闭连接
    ssh.close()


remote_connect()

2.2 连接方式二：

定义一个linux类实现连接远程连接linux服务器

import paramiko
import re
from time import sleep
# 定义一个类，表示一台远端linux主机
class Linux(object):
    # 通过IP, 用户名，密码，超时时间初始化一个远程Linux主机
    def __init__(self, ip, username, password, timeout=30):
        self.ip = ip
        self.username = username
        self.password = password
        self.timeout = timeout
        # transport和chanel
        self.t = ''
        self.chan = ''
        # 链接失败的重试次数
        self.try_times = 3
 
    # 调用该方法连接远程主机
    def connect(self):
        while True:
            # 连接过程中可能会抛出异常，比如网络不通、链接超时
            try:
                self.t = paramiko.Transport(sock=(self.ip, 22))
                self.t.connect(username=self.username, password=self.password)
                self.chan = self.t.open_session()
                self.chan.settimeout(self.timeout)
                self.chan.get_pty()
                self.chan.invoke_shell()
                # 如果没有抛出异常说明连接成功，直接返回
                print(u'连接%s成功' % self.ip)
                # 接收到的网络数据解码为str
                print(self.chan.recv(65535).decode('utf-8'))
                return
            # 这里不对可能的异常如socket.error, socket.timeout细化，直接一网打尽
            except Exception:
                if self.try_times != 0:
                    print(u'连接%s失败，进行重试' % self.ip)
                    self.try_times -= 1
                else:
                    print(u'重试3次失败，结束程序')
                    exit(1)
 
    # 断开连接
    def close(self):
        self.chan.close()
 
        self.t.close()
 
    # 发送要执行的命令
    def send(self, cmd,pattern):
        cmd += '\\r'
        # 通过命令执行提示符来判断命令是否执行完成
        patt = pattern
        p = re.compile(patt)
        result = ''
        # 发送要执行的命令
        self.chan.send(cmd)
        # 回显很长的命令可能执行较久，通过循环分批次取回回显
        while True:
            sleep(0.5)
            ret = self.chan.recv(65535)
            ret = ret.decode('utf-8')
            result += ret
            if p.search(ret):
                print(result)
                return result
 
#测试linux类代码
# if __name__ == '__main__':
#      host = Linux('192.168.1.106', 'long', 'long')
#      host.connect()
#      host.send('su root',r'.*(]#|密码).*')
#      host.send('123456')
#      host.send('pwd')
#      host.close()

2.3 连接方式三：

定义一个App类调用linux类根据业务需求传入不同的参数实现业务不同的功能。（此时linux类作为一个工具类）

from common.connectionLinux.Linux import Linux
 
class App(object):
    #主节点服务器IP地址
    MASTIP = '192.168.1.106'
    #多个服务器IP地址
    IPlist = ['192.168.1.107','192.168.1.108','192.168.1.110']
 
    def __init__(self):
        self.host = Linux(self.MASTIP, 'long', 'long')
    #非root账户登录
    def connect(self):
        self.host.connect()
 
    #切换root账户登录
    def sendCmdRoot(self):
        pattern1 = r'.*(密码).*'
        pattern2 = r'.*(]#).*'
        cmd1 = 'su root'
        cmd2 = '123456'
        self.host.send(cmd1,pattern1)
        self.host.send(cmd2,pattern2)
 
    #子节点文件拷贝到主节点
    def copyCmd(self):
        for ip in self.IPlist:
            print(ip)
 
        pattern = r'.*(]#).*'
        cmd = 'scp root@'+ip+':/home/long/test.txt /home/long'
        print(cmd)
        self.host.send(cmd,pattern)
 
 
if __name__ == '__main__':
    #连接linux
    app = App()
    # app.connect()
    # app.sendCmdRoot()
    app.copyCmd()