Python之paramiko远程控制工具

paramiko基本使用
ssh是一个协议，OpenSSH是其中一个开源实现，paramiko是Python的一个库，实现了SSHv2协议(底层使用cryptography)。
有了Paramiko以后，我们就可以在Python代码中直接使用SSH协议对远程服务器执行操作，而不是通过ssh命令对远程服务器进行操作。
Transport:加密会话

connect(self,hostname,port=22,username=None,password=None,pkey=None,key_filename=None,timeout=None,allow_agent=True,look_for_keys=True,compress=False)

参数说明：
pkey:私钥方式用户验证
key_filename:私钥文件名
timeout:连接超时时间
allow_agent:是否允许ssh代理
look_for_keys:是否需要搜索私钥文件
compress:打开时压缩

基本框架：

import paramiko

# 实例化SSHClient
client = paramiko.SSHClient()

# 自动添加策略，保存服务器的主机名和密钥信息，如果不添加，那么不再本地know_hosts文件中记录的主机将无法连接
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

# 连接SSH服务端，以用户名和密码进行认证
client.connect(hostname=\'127.0.0.1\', port=22, username=\'root\', password=\'123456\')

# 打开一个Channel并执行命令
stdin, stdout, stderr = client.exec_command(\'df -h \')  # stdout 为正确输出，stderr为错误输出，同时是有1个变量有值

# 打印执行结果
print(stdout.read().decode(\'utf-8\'))

# 关闭SSHClient
client.close()

实例一、

远程执行脚本

#!/usr/bin/python
# -*- coding:utf-8 -*-
host_list = [\'127.0.0.1\']
port = 22
username = \'zhoujt\'
password = \'1234\'
for host in host_list:
    import paramiko
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    client.connect(hostname=host,port=22,username=username,password=password)
    print(\'The connect success!\')
    stdin, stdout, stderr = client.exec_command(\'sudo python /home/zhoujt/ceshi/ceshi.py\', get_pty=True)
    #sudo 休眠一秒来输入密码
    time.sleep(1)
    stdin.write(\'1234\n\')
    print(stdout.read().decode(\'utf-8\'))
    client.close()

实例二、

由于实例一的密码是明文推送，这里使用getpass来传输密文格式的字符。

# -*- coding:utf-8 -*-
import paramiko
import time
import getpass

ip = input(\'input your ip :\')
username = input(\'Username : \')
password = getpass.getpass(\'Input your password :\')
# print(password)
ssh_client = paramiko.SSHClient()
ssh_client.set_missing_host_key_prolicy(paramiko.AutoAddPolicy())
ssh_client.connect(hostname=ip, username=username, password=password, look_for_keys=False)
print("You have successfull connect to", ip)
command = paramiko.client.invoke_shell()
cmdlist = open(\'D:\\Python\\python01\\cmd_file\', \'r\')
cmdlist.seek(0)
print(cmdlist.seek(0))
for line in cmdlist.readlines():
command.send(line + "\n")
time.sleep(1)
cmdlist.close()
output = command.recv(65535).decode(\'ASCII\')
print(output)
ssh_client.close

实例三、

通过Stfp来实现文件的传输

host_list = [\'1.1.1.1\',
\'2.2.2.2\'    ,
\'3.3.3.3\'    ]
port = 22
username = \'zhoujt\'
password = \'password\'

for host in host_list:
    import paramiko
    transport = paramiko.Transport(host, port)
    transport.connect(username=username, password=password)
    sftp = paramiko.SFTPClient.from_transport(transport)
    # 将location.py 上传至服务器 /tmp/test.py
    sftp.put(\'/tmp/location.py\', \'/tmp/test.py\')
    # 将remove_path 下载到本地 local_path
    sftp.get(\'/home/zhoujt/zhoujt.py\', \'D:\\Python\\zhoujt.py\')
    transport.close()
    print(host,\' 已传输完成!\')

 

实例四、

脚本常用的时间

#!/usr/bin/python
# coding=utf-8
import datetime

# 获取当前时间
now = datetime.datetime.now()
# 获取今天零点
zeroToday = now - datetime.timedelta(hours=now.hour, minutes=now.minute, seconds=now.second,microseconds=now.microsecond)
# 获取23:59:59
lastToday = zeroToday + datetime.timedelta(hours=23, minutes=59, seconds=59)
# 获取前一天的当前时间
yesterdayNow = now - datetime.timedelta(hours=23, minutes=59, seconds=59)
# 获取明天的当前时间
tomorrowNow = now + datetime.timedelta(hours=23, minutes=59, seconds=59)

print(\'时间差\', datetime.timedelta(hours=23, minutes=59, seconds=59))
print(\'当前时间\', now)
print(\'今天零点\', zeroToday)
print(\'获取最后一秒\', lastToday)
print(\'昨天当前时间\', yesterdayNow)
print(\'明天当前时间\', tomorrowNow)
# 输出：
# 时间差 23:59:59
# 当前时间 2021-04-28 09:15:46.938364
# 今天零点 2021-04-28 00:00:00
# 获取最后一秒 2021-04-28 23:59:59
# 昨天当前时间 2021-04-27 09:15:47.938364
# 明天当前时间 2021-04-29 09:15:45.938364

实例五、功能合集

class SSHConnection(object):
 
    def __init__(self, host_dict):
        self.host = host_dict[\'host\']
        self.port = host_dict[\'port\']
        self.username = host_dict[\'username\']
        self.pwd = host_dict[\'pwd\']
        self.__k = None
 
    def connect(self):
        transport = paramiko.Transport((self.host,self.port))
        transport.connect(username=self.username,password=self.pwd)
        self.__transport = transport
 
    def close(self):
        self.__transport.close()
 
    def run_cmd(self, command):

        ssh = paramiko.SSHClient()
        ssh._transport = self.__transport
        stdin, stdout, stderr = ssh.exec_command(command)
        res = unicode_utils.to_str(stdout.read())
        # 打印错误信息
        error = unicode_utils.to_str(stderr.read())
        如果报错则
        if error.strip():
            return {\'color\':\'red\',\'res\':error}
        else:
            return {\'color\': \'green\', \'res\':res}
 
    def upload(self,local_path, target_path):
        sftp = paramiko.SFTPClient.from_transport(self.__transport)
        sftp.put(local_path, target_path, confirm=True)
        # print(os.stat(local_path).st_mode)
        # 增加权限
        # sftp.chmod(target_path, os.stat(local_path).st_mode)
        # 注意这里的权限是八进制的，八进制需要使用0o作为前缀
        sftp.chmod(target_path, 0o755) 
 
    def download(self,target_path, local_path):
        # 连接，下载
        sftp = paramiko.SFTPClient.from_transport(self.__transport)
        # 将location.py 下载至服务器 /tmp/test.py
        sftp.get(target_path, local_path)
 
    def __del__(self):
        self.close()
 
　　
#unicode_utils.py
def to_str(bytes_or_str):
    """
    把byte类型转换为str
    :param bytes_or_str:
    :return:
    """
    if isinstance(bytes_or_str, bytes):
        value = bytes_or_str.decode(\'utf-8\')
    else:
        value = bytes_or_str
    return value

 

详细请参考以下博客：

python模块学习 - paramiko - 一只小小的寄居蟹 - 博客园

python的paramiko模块 - breezey - 博客园