python使用paramiko执行服务器脚本并拿到实时结果
作者:LanLanDeMing 发布时间:2021-06-02 17:54:29
标签:python,paramiko,服务器,脚本,实时结果
paramiko 执行服务器脚本并拿到实时结果
import paramiko
cmd = '{0}/{1} linux 32'.format('/root/installer', 'make_client_installer.sh')
print(cmd)
try:
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect('xx.xx.xx.xx', port, 'username', 'password', timeout=5)
stdin, stdout, stderr = ssh.exec_command(cmd, get_pty=True)
while not stdout.channel.exit_status_ready():
result = stdout.readline()
print(result)
if stdout.channel.exit_status_ready():
a = stdout.readlines()
print(a)
break
ssh.close()
except Exception as e:
print(e)```
python paramiko模块使用
paramiko远程密码连接
# 基于ssh用于连接远程服务器做操作:远程执行命令, 上传文件, 下载文件
import paramiko
# ssh root@172.25.254.250
# 创建一个ssh对象;
client = paramiko.SSHClient()
# 2. 解决问题:如果之前没有;连接过的ip, 会出现
# Are you sure you want to continue connecting (yes/no)? yes
# 自动选择yes
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
# 3. 连接服务器
client.connect(hostname='172.25.254.1',
port=22,
username='root',
password='redhat')
# 4. 执行操作
stdin, stdout, stderr = client.exec_command('hostnaewdeme')
# 5. 获取命令的执行结果;
result = stdout.read().decode('utf-8')
print(result)
print(stderr.read())
# 6. 关闭连接
client.close()
paramiko批量远程密码连接
# 基于ssh用于连接远程服务器做操作:远程执行命令, 上传文件, 下载文件
import paramiko
import logging
from paramiko.ssh_exception import NoValidConnectionsError, AuthenticationException
def connect(cmd, hostname, port=22, username='root', password='westos'):
# ssh root@172.25.254.250
# 创建一个ssh对象;
client = paramiko.SSHClient()
# 2. 解决问题:如果之前没有;连接过的ip, 会出现
# Are you sure you want to continue connecting (yes/no)? yes
# 自动选择yes
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
try:
# 3. 连接服务器
client.connect(hostname=hostname,
port=port,
username=username,
password=password)
print("正在连接主机%s......." %(hostname))
except NoValidConnectionsError as e:
print("连接失败")
except AuthenticationException as e:
print("密码错误")
else:
# 4. 执行操作
stdin, stdout, stderr = client.exec_command(cmd)
# 5. 获取命令的执行结果;
result = stdout.read().decode('utf-8')
print(result)
# 6. 关闭连接
client.close()
with open('host.txt') as f:
for line in f:
line = line.strip()
hostname, port, username, password = line.split(':')
print(hostname.center(50, '*'))
connect('hostname', hostname, port, username, password)
paramiko基于公钥密钥连接
# 基于ssh用于连接远程服务器做操作:远程执行命令, 上传文件, 下载文件
import paramiko
from paramiko.ssh_exception import NoValidConnectionsError, AuthenticationException
def connect(cmd, hostname, port=22, user='root'):
# ssh root@172.25.254.250
# 创建一个ssh对象;
client = paramiko.SSHClient()
# 返回一个私钥对象
private_key = paramiko.RSAKey.from_private_key_file('id_rsa')
# 2. 解决问题:如果之前没有;连接过的ip, 会出现
# Are you sure you want to continue connecting (yes/no)? yes
# 自动选择yes
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
try:
# 3. 连接服务器
client.connect(hostname=hostname,
port=port,
username=user,
pkey=private_key
)
# 4. 执行操作
stdin, stdout, stderr = client.exec_command(cmd)
except NoValidConnectionsError as e:
print("连接失败")
except AuthenticationException as e:
print("密码错误")
else:
# 5. 获取命令的执行结果;
result = stdout.read().decode('utf-8')
print(result)
finally:
# 6. 关闭连接
client.close()
for count in range(254):
host = '172.25.254.%s' %(count+1)
print(host.center(50, '*'))
connect('uname', host)
基于用户名密码上传下载
import paramiko
transport = paramiko.Transport(('172.25.254.39', 22))
transport.connect(username='root', password='westos')
sftp = paramiko.SFTPClient.from_transport(transport)
# 上传文件, 包含文件名
sftp.put('/tmp/kiosk', '/mnt/kiosk1')
sftp.get('/mnt/kiosk', '/home/kiosk/Desktop/day18/kiosk')
transport.close()
基于密钥上传下载
import paramiko
# 返回一个私钥对象
private_key = paramiko.RSAKey.from_private_key_file('id_rsa')
transport = paramiko.Transport(('172.25.254.39', 22))
transport.connect(username='root',pkey=private_key)
sftp = paramiko.SFTPClient.from_transport(transport)
# 上传文件, 包含文件名
sftp.put('/tmp/kiosk', '/mnt/kiosk2')
sftp.get('/mnt/kiosk2', '/home/kiosk/Desktop/day18/kiosk')
transport.close()
paramiko再次封装
import os
import paramiko
from paramiko.ssh_exception import NoValidConnectionsError, AuthenticationException, SSHException
class SshRemoteHost(object):
def __init__(self, hostname, port, user, passwd, cmd):
# 指的不是shell命令
# cmd shell命令
# put
# get
self.hostname = hostname
self.port = port
self.user = user
self.passwd = passwd
self.cmd = cmd
def run(self):
"""默认调用的内容"""
# cmd hostname
# put
# get
cmd_str = self.cmd.split()[0] # cmd
# 类的反射, 判断类里面是否可以支持该操作?
if hasattr(self, 'do_'+ cmd_str): # do_cmd
getattr(self, 'do_'+cmd_str)()
else:
print("目前不支持该功能")
def do_cmd(self):
# 创建一个ssh对象;
client = paramiko.SSHClient()
# 2. 解决问题:如果之前没有;连接过的ip, 会出现
# Are you sure you want to continue connecting (yes/no)? yes
# 自动选择yes
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
try:
# 3. 连接服务器
client.connect(hostname=self.hostname,
port=self.port,
username=self.user,
password=self.passwd)
print("正在连接主机%s......." % (self.hostname))
except NoValidConnectionsError as e:
print("连接失败")
except AuthenticationException as e:
print("密码错误")
else:
# 4. 执行操作
# cmd uname
# cmd ls /etc/
# *******注意:
cmd = ' '.join(self.cmd.split()[1:])
stdin, stdout, stderr = client.exec_command(cmd)
# 5. 获取命令的执行结果;
result = stdout.read().decode('utf-8')
print(result)
# 6. 关闭连接
client.close()
def do_put(self):
# put /tmp/passwd /tmp/passwd
# put /tmp/passwd /tmp/pwd
# put /tmp/passwd # 将本机的/tmp/passwd文件上传到远程主机的/tmp/passwd;
print("正在上传.....")
try:
transport = paramiko.Transport((self.hostname, int(self.port)))
transport.connect(username=self.user, password=self.passwd)
except SSHException as e:
print("连接失败")
else:
sftp = paramiko.SFTPClient.from_transport(transport)
newCmd = self.cmd.split()[1:]
if len(newCmd) == 2:
# 上传文件, 包含文件名
sftp.put(newCmd[0], newCmd[1])
print("%s文件上传到%s主机的%s文件成功" %(newCmd[0],
self.hostname, newCmd[1]))
else:
print("上传文件信息错误")
transport.close()
def do_get(self):
print("正在下载.....")
# 2. 根据选择的主机组, 显示包含的主机IP/主机名;
# 3. 让用户确认信息, 选择需要批量执行的命令;
# - cmd shell命令
# - put 本地文件 远程文件
# - get 远程文件 本地文件
def main():
# 1. 选择操作的主机组:eg: mysql, web, ftp
groups = [file.rstrip('.conf') for file in os.listdir('conf')]
print("主机组显示:".center(50, '*'))
for group in groups: print('\t', group)
choiceGroup = input("清选择批量操作的主机组(eg:web):")
# 2. 根据选择的主机组, 显示包含的主机IP/主机名;
# 1). 打开文件conf/choiceGroup.conf
# 2). 依次读取文件每一行,
# 3). 只拿出ip
print("主机组包含主机:".center(50, '*'))
with open('conf/%s.conf' %(choiceGroup)) as f:
for line in f:
print(line.split(':')[0])
f.seek(0,0) # 把指针移动到文件最开始
hostinfos = [line.strip() for line in f.readlines()]
# 3. 让用户确认信息, 选择需要批量执行的命令;
print("批量执行脚本".center(50, '*'))
while True:
cmd = input(">>:").strip() # cmd uname
if cmd:
if cmd == 'exit' or cmd =='quit':
print("执行结束, 退出中......")
break
# 依次让该主机组的所有主机执行
for info in hostinfos:
# 'ip:port:user:passwd'
host, port, user, passwd = info.split(":")
print(host.center(50, '-'))
clientObj = SshRemoteHost(host, port, user, passwd, cmd)
clientObj.run()
if __name__ == '__main__':
main()
来源:https://blog.csdn.net/LanlanDeming/article/details/113700202


猜你喜欢
- 本文实例讲述了Python正则替换字符串函数re.sub用法。分享给大家供大家参考,具体如下:python re.sub属于python正则
- golang判断元素是否在数组内众所周知,golang里没有像python的in来判断元素是否在list里存在,可替代的办法是将list放到
- YSlow是yahoo美国开发的一个页面评分插件,非常的棒,从中我们可以看出我们页面上的很多不足,并且可以知道我们改怎么却改进和优化。仔细研
- 测试环境 硬件:CPU 酷睿双核T5750 内存:2G 软件:Windows server 2003 + sql server 2005 O
- 前言:在Vue项目中,当访问的页面路由不存在或错误时,页面显示为一片空白。然而,通常我们需要对访问url不存在或者错误的情况下添加默认的40
- 事件调度sched 模块内容很简单,只定义了一个类。它用来最为一个通用的事件调度模块。class sched.scheduler(timef
- 如何在ADO中使用存储查询?对于使用参数存贮查询,我们可用下面的代码进行示例:Private cn As Ne
- 下边的函数,实现了象数组一样去处理字符串。 一,用临时表作为数组 create function f_split(@c varchar(20
- 不同数据库之间若不能导入导出,那么将是一件可怕的事情,所幸的是一般情况下通过不同的方法和途径,都可以实现,方法有多种,本人提供其中的一个,提
- 用requests包请求https的网站时,我们偶尔会遇到证书问题。也就是常见的SSLerror,遇到这种问题莫慌莫慌。这里没有找到合适的网
- 本文实例为大家分享了小程序实现横向滑动日历效果的具体代码,供大家参考,具体内容如下<scroll-view class="s
- 1.条件语句几个注意点和C#不一样的。if a < 5 { return 0} else {
- 为了降低用户注册难度,国际站的主注册表单一直在改进。主注册三月至今发生了两次较大的变化,现在对表单调整的地方分解如下:1. 两步
- Python编写类的时候,每个函数参数第一个参数都是self,一开始我不管它到底是干嘛的,只知道必须要写上。后来对Python渐渐熟悉了一点
- 问题怎样捕获代码中的所有异常?解决方案想要捕获所有的异常,可以直接捕获 Exception 即可:try: ...except
- 本文实例讲述了Python实现的随机森林算法。分享给大家供大家参考,具体如下:随机森林是数据挖掘中非常常用的分类预测算法,以分类或回归的决策
- 输入汉字提示拼音,试试下面这个函数,不知是不是你要的那个:查询汉字便宜到词典网<%function getpychar(ch
- antd-日历组件,前后禁止选择,只能选中间一部分:dateDisabledDate(current) { // 需求有效期的禁止选择时间
- 正常开发的时候,我们都手动去写要引入到包,有过java开发的同事,用过快捷键ctrl + alt + o 会自动引入所有的依赖包,pycha
- php去掉数组的第一个值的两种删除方法:1、使用array_shift()函数删除数组的第一个值,语法“array_sh