ssh操作及接受响应,自动获取test密码
2023-06-13 09:11:25 时间
一、ssh操作及接受响应
# coding=utf-8
"""
@Project :pachong-master
@File :common.py
@Author :gaojs
@Date :2022/7/9 14:01
@Blogs : https://www.gaojs.com.cn
"""
import sys
import logging
import time
import paramiko
from time import sleep
import re
import requests
from faker import Factory
sys.setrecursionlimit(5000)
class CLI:
def ssh_ag(self, hostname='192.168.120.209', port=22, username='array', password='admin'):
"""
:param self:
:return:
"""
# 创建ssh对象
self.fin = open('log.txt', 'w')
self.ssh = paramiko.SSHClient()
self.logging = logging
# 允许连接不在know_hosts文件中的主机
self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy)
# 连接AG
self.ssh.connect(hostname=hostname, port=port, username=username, password=password)
sleep(5)
channel = self.ssh.invoke_shell()
self.channel = channel
channel.settimeout(5)
output = channel.recv(2048).decode('utf-8')
time.sleep(1)
self.fin.write(output)
self.cli_cmd('enable')
self.cli_cmd('')
self.cli_cmd('config ter')
self.cli_cmd('no page')
def print_step(self):
"""
:return:
"""
result = self.channel.recv(2048)
print(result.decode())
def cli_cmd(self, cli, prompt="[#\$]", timeout=3):
"""
:param cli:
:return:
"""
output = ''
self.logging.info("AG execute command in enable mode: " + cli)
self.channel.send(cli + '\n')
output = self.read_until(prompt, timeout)
return output
def quit_enable(self):
"""
:return:
"""
self.cli_cmd('switch global')
self.cli_cmd('exit')
def clear_log(self):
"""
:return:
"""
sleep(2)
self.cli_cmd('clear log b')
def read_until(self, expected, timeout=10):
"""
等待回显:这个方法是从同事那里偷来的哈哈哈
:return:
"""
output = ''
regexp = re.compile(expected)
max_time = time.time() + timeout
i = 0
while time.time() < max_time:
i = i + 1
tmp = ""
try:
tmp = self.channel.recv(1024).decode('utf-8')
except:
pass
self.fin.write(tmp)
self.fin.flush()
output += tmp
if regexp.search(output):
return output
return output
def switch_vsite(self, vsite):
"""
切换虚拟站点
:param vsite:
:return:
"""
self.cli_cmd('sw %s' % vsite)
def get_sn(self):
"""
获取sn码
:return:
"""
# with open('log.txt', mode='r') as fin:
# resp = fin.read()
resp = self.cli_cmd('show version')
result = re.compile('Serial Number : [A-Z0-9]{31}')
sn = result.findall(resp)[0]
sn_number = sn.split(':')[1]
# print(sn_number)
return sn_number
def create_test_password(self, sn):
"""
生成test用户密码
"""
url = 'http://10.3.0.50/cgi-bin/passwd_res'
f = Factory.create()
ua = f.user_agent()
headers = {
'User-Agent': ua
}
data = {
'serial': sn
}
rsp = requests.post(url=url, headers=headers, data=data)
passwd = re.findall('password: (.*?)</pre>', rsp.text)[0]
print(passwd)
return passwd
if __name__ == '__main__':
"""
自动获取test密码
"""
ag_ip = input('请输入您AG管理IP:')
test = CLI()
test.ssh_ag(hostname=ag_ip)
sn_number = test.get_sn()
passwd = test.create_test_password(sn_number)
print(f'*********************** 您的test账户密码是 {passwd} ************************\n')
2.云服务器操作
# coding=utf-8
"""
@Project :pachong-master
@File :test01.py
@Author :gaojs
@Date :2022/7/9 15:40
@Blogs : https://www.gaojs.com.cn
"""
import sys
import logging
import time
import paramiko
from time import sleep
import re
sys.setrecursionlimit(5000)
class CLI:
def ssh_ag(self, hostname='101.4x.39.xxx', port=22, username='root', password='xxxxxx'):
"""
:param self:
:return:
"""
# 创建ssh对象
self.fin = open('log.txt', 'w+')
self.ssh = paramiko.SSHClient()
self.logging = logging
# 允许连接不在know_hosts文件中的主机
self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy)
# 连接AG
self.ssh.connect(hostname=hostname, port=port, username=username, password=password)
sleep(5)
channel = self.ssh.invoke_shell()
self.channel = channel
channel.settimeout(5)
output = channel.recv(2048).decode('utf-8')
time.sleep(1)
self.fin.write(output)
def print_step(self):
"""
:return:
"""
result = self.channel.recv(2048)
print(result.decode())
def cli_cmd(self, cli, prompt="[#\$]", timeout=3):
"""
:param cli:
:return:
"""
output = ''
self.logging.info("cloud server execute command in enable mode: " + cli)
self.channel.send(cli + '\n')
output = self.read_until(prompt, timeout)
return output
def read_until(self, expected, timeout=10):
"""
等待回显
:return:
"""
output = ''
regexp = re.compile(expected)
max_time = time.time() + timeout
i = 0
while time.time() < max_time:
i = i + 1
tmp = ""
try:
tmp = self.channel.recv(1024).decode('utf-8')
except:
pass
self.fin.write(tmp)
self.fin.flush()
output += tmp
if regexp.search(output):
return output
return output
if __name__ == '__main__':
"""
自动获取test密码
"""
test = CLI()
test.ssh_ag()
test.cli_cmd('ll /opt')
test.cli_cmd('cd /var/www/html/')
test.cli_cmd('date')
test.cli_cmd('ls')
相关文章
- 【Jenkins 插件】使用 SSH Slaves 创建从节点执行任务
- SSH Config 允许使用root密码登陆 PermitRootLogin[通俗易懂]
- 使用lrzsz上传ssh客户端公钥[通俗易懂]
- Ubuntu20.04基础:桌面安装及root权限开通和ssh安装讲解
- python 通过ssh 连接Linux系统并发送指令详解程序员
- Maven搭建SSH(Struts2+Spring+Hibernate)框架入门教程(附带实例)
- 快速连接Linux:SSH远程管理工具(ssh连接linux工具)
- 分析利用Linux SSH日志精准分析入侵行为(linuxssh日志)
- 登录Linux远程配置SSH安全登录(linux远程配置ssh)
- 使用 Linux 系统设置 SSH 客户端连接(ssh客户端linux)
- Linux查看SSH版本的方法:简单快捷(linux查看ssh版本)
- ssh 登录 Linux远程登录无需秘钥:SSH认证更安全(linux免秘钥)
- Linux 如何使用SSH进行网络翻墙(linux如何ssh翻墙)
- 使用SSH连接Oracle数据库(ssh连接oracle)
- 管理你的Linux SSH服务器(linuxssh服务器)
- 如何在CentOS/RHEL上设置SSH免密码登录
- 展示SSH连接Linux应用的精彩之旅(ssh连接linux软件)
- 利用Oracle SSH隧道安全远程访问(oracle ssh隧道)