正文
python 主机宝
提示：扫一扫查出行【扫一扫了解最新限行尾号】
复制提示
需求：开发一个主机批量管理系统，要求按saltstack方式执行命令
#!/usr/bin/env python3.
# -*- coding:utf8 -*-
import os,sys,pickle,logging
BASEDIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.append(BASEDIR)
from conf import setting
from core import file_handler
from core import db_handler
from core import host_handler
"""
************************************
此为主机宝主运行程序
************************************
"""
logging.basicConfig(level=logging.INFO, filename=os.path.join(BASEDIR,'log/ssh.log'), filemode='a',
format='%(asctime)s %(levelname)s %(message)s', datefmt='%Y/%m/%d %H:%M:%S')
log = logging.getLogger(__name__)
def login():
count =
flage = False
while count < :
count +=
user_input = input("请输入用户名：").strip()
pass_input = input("请输入密码：").strip()
db = db_handler.handler(setting.DATABASE,user_input)
if os.path.isfile(db):
f = open(db,"rb")
data = pickle.loads(f.read())
f.close()
if user_input == data["name"] and data["lock"] !=:
if pass_input == data["password"]:
flage = True
log.info("用户[%s]登陆成功！"%user_input)
break
else:
print("用户名或密码错误！")
if count > :
with open(db,"wb") as f:
data["lock"] =
pickle.dump(data,f)
log.info("用户[%s]被锁定！"%user_input)
print("用户[%s]已被锁定！"%user_input)
else:
print("用户[%s]已被锁定！"%user_input)
exit()
if flage == True:
print("用户[%s]登陆成功！"%user_input)
men()
else:
exit()
def men():
print("欢迎进入主机宝管理系统!")
host_men = """
、显示主机与所属组
、增加组
、增加主机
、修改主机
、删除主机
、执行命令
、退出管理系统
"""
host_dic ={
"":{"option":"显示主机与所属组","action":file_handler.show},
"":{"option":"增加组","action":file_handler.add_group},
"":{"option":"增加主机","action":file_handler.add_host},
"":{"option":"修改主机","action":file_handler.mod_host},
"":{"option":"删除主机","action":file_handler.host_delete},
"":{"option":"执行命令","action":host_handler.exciton},
"":{"option":"退出管理系统","action":exit}
}
exit_flag =False
while not exit_flag:
print(host_men)
option = input("请按键选择：").strip()
if option in host_dic:
func = host_dic[option].get("action")
func() def run():
login()
main
#!/usr/bin/env python3.5
# -*- coding:utf8 -*-
import os,sys,re
BASEDIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.append(BASEDIR)
import logging,importlib
# 初始化日志格式及对象
logging.basicConfig(level=logging.INFO, filename=os.path.join(BASEDIR,'log/ssh.log'), filemode='a',
format='%(asctime)s %(levelname)s %(message)s', datefmt='%Y/%m/%d %H:%M:%S')
log = logging.getLogger(__name__)
# 调用执行模块
def module_excute(moudle_name,func_name,arg):
try:
# 导入要执行的模块
module = importlib.import_module("salt.salt_{}".format(moudle_name))
# 判断函数名是否包含在模块里
if hasattr(module,func_name):
func = getattr(module,func_name)
func(arg)
print("***" * 20)
else:
print("不存在")
except Exception as e :
log.info("input:{},error:{}".format(moudle_name,e))
def exciton():
usage = """
salt "*" cmd.run "excute_cmd1,excute_cmd2..." :"所有主机执行命令"
salt -g "group" cmd.run "excute_cmd1,excute_cmd2..." :"指定组执行命令"
salt -h "ip_host" cmd.run "excute_cmd1,excute_cmd2..." :"指定主机IP执行命令"
salt "*" file.put "filename" :"所有主机上传文件"
salt "*" file.get "filename" :"所有主机下载文件"
exit :"退出"
"""
print("欢迎进入主机命令执行系统!")
user_cmd = input("请输入要执行的命令>>>:").strip()
if user_cmd.startswith("salt"): # 判断是否以salt开始
user_cmd_list = user_cmd.split() #以空格分割成列表
# 过滤掉特殊字符
user_arg_list = list(map(lambda x:re.sub(r'[\"\']',"",x),user_cmd_list))
# 匹配含点的模块名字
p = re.compile(r'[a-zA-Z_]+\.[a-zA-Z_]+')
flag =False
count = 0
for i in user_arg_list:
if p.match(i):
flag = True
count +=1
moudle_func = i # 获取模块名
break # 只匹配第一个含点的模块名
# 只有命令里含*。*格式时，继续
if flag and count == 1:
cmd_list = user_arg_list[user_arg_list.index(moudle_func)+1:] # 获取原列表在此命令（*.*）之后的所有命令变成命令列表
obj_list = user_arg_list[user_arg_list.index("salt")+1:user_arg_list.index(moudle_func)] # 获取以salt开头模块函数结尾之前的所有内空转到列表
arg = (obj_list,cmd_list) # 将操作对象列表和指令列表放到元组中
moudle_name = moudle_func.split(".")[0] # 获取模块名
func_name = moudle_func.split(".")[1] # 获取函数名
module_excute(moudle_name,func_name,arg)
exciton()
else:
print("命令输入错误！请按以下格式输入：")
print(usage)
exciton()
elif user_cmd =="exit":
exit()
else:
print("命令输入错误！请按以下格式输入：")
print(usage)
exciton()
host_handle
#!/usr/bin/env python3.5
# -*- coding:utf8 -*-
import os,sys,pickle,re,logging
BASEDIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.append(BASEDIR)
from conf import setting
from core import db_handler
from core import host_handler
db_path = db_handler.handler(setting.DATABASE,"host")
if os.path.exists(db_path):
with open(db_path, "rb") as f:
data = pickle.loads(f.read())
else:
data =[]
# 初始化日志格式及对象
logging.basicConfig(level=logging.INFO, filename=os.path.join(BASEDIR,'log/ssh.log'), filemode='a',
format='%(asctime)s %(levelname)s %(message)s', datefmt='%Y/%m/%d %H:%M:%S')
log = logging.getLogger(__name__)
def check_ip(ip):
for data_ip in data:
for ip_data in data_ip:
for iptest in data_ip[ip_data]:
if ip in iptest["ip"]:
return True
else:
return False
def check_group(group):
for gc in data:
if group in gc.keys():
return True
else:
return False
def add_group():
add_input = input("请输入要增加的组：").strip()
list_data = []
for y_data in data:
for k in y_data:
list_data.append(k)
if add_input not in list_data:
new_group = {"%s"%add_input:[]}
data.append(new_group)
with open(db_path,"wb") as fw:
pickle.dump(data,fw)
log.info("增加组%s成功！"%add_input)
print("增加组%s成功！"%add_input)
else:
log.error("增加组%s失败！已存在该组!"%add_input)
print("增加组%s失败！已存在该组!"%add_input)
def show():
for y_data in data:
for k in y_data:
for i in y_data[k]:
print("主机IP：[%s],所属组为：[%s]"%(i["ip"],k))
def add_host():
"""
增加主机
:return:
"""
try:
host_add = input("请输入主机IP：").strip()
host_port = int(input("请输入端口号："))
host_user = input("请输入登陆主机用户名：").strip()
host_pwd = input("请输入登陆主机密码：").strip()
# 判断是否为IP
if re.match(r"((2[0-4]\d|25[0-5]|[01]?\d\d?)\.){3}(2[0-4]\d|25[0-5]|[01]?\d\d?)$",host_add):
if host_port != "" and len(host_user) != 0 and len(host_pwd)!= 0:
host_group = input("请输入主机所属组:").strip()
for g in data:
if host_group in g.keys():
for g_data in data:
if host_group in g_data.keys():
g_data[host_group].append({"ip":"%s"%host_add,"port":"%s"%host_port,"username":"%s"%host_user,"password":"%s"%host_pwd})
with open(db_path,"wb") as fw:
pickle.dump(data,fw)
log.info("增加主机[%s]成功！"%host_add)
print("增加主机[%s]成功！"%host_add)
break
else:
log.error("增加主机[%s]失败,组[%]不存在！"%(host_add,host_group))
print("增加主机[%s]失败，组[%]不存在！！"%(host_add,host_group))
return add_host()
else:
return add_host()
else:
log.error("你输入的不是IP地址：%s" %host_add)
print("你输入的不是IP地址：%s" %host_add)
except Exception as ex:
log.error("增加主机异常%s"%ex)
print("增加主机异常")
def mod_host():
"""
修改主机所属组
:return:
"""
IP_modi = input("请输入要变更的IP：").strip()
ip_check = check_ip(IP_modi)
if ip_check:
gg = input("请输入转入的组名称：").strip()
gg_check = check_group(gg)
if gg_check:
# 获取该IP 原所属组名称
for data_gg in data:
for i_gg in data_gg:
for i,ip_data in enumerate(data_gg[i_gg]):
if IP_modi == ip_data["ip"]:
g = i_gg
count = i
ip = ip_data
if gg == g:
log.info("该IP主机：{}，原已属于该组：{}".format(IP_modi,gg))
print("该IP主机：{}，原已属于该组：{}".format(IP_modi,gg))
else:
for x_data in data:
for xi_gg in x_data:
# 确定转入组相符
if xi_gg == gg:
x_data[xi_gg].append(ip)
# 删除原来所属组IP主机
elif xi_gg == g:
x_data[xi_gg].remove(ip)
with open(db_path,"wb") as fw:
pickle.dump(data,fw)
log.info("修改主机[%s]成功,新组名称为%s！"%(IP_modi,gg))
print("修改主机[%s]成功,新组名称为%s！"%(IP_modi,gg))
else:
log.error("不存在此IP主机{}".format(IP_modi))
print("不存在此IP主机{}".format(IP_modi))
def cmd_handle(arg):
"""
解析命令，并返回主机IP列表
:param arg:
:return:
"""
if arg[0] == "*":
ip_list = []
for g in data:
for gg in g:
for ip in g[gg]:
ip_list.append(ip["ip"])
ip_list = list(set(ip_list)) # 去除重复IP
return ip_list
elif arg[0] == "-h":
ip_list=[]
ip_group = arg[1:]
for data_ip in data:
for ip in ip_group:
for ip_data in data_ip:
for iptest in data_ip[ip_data]:
if ip in iptest["ip"]:
ip_list.append(ip)
ip_list =list(set(ip_list))
return ip_list
elif arg[0] == "-g":
ip_list = []
group_list =arg[1:]
for group in group_list:
for g in data:
if group in g.keys():
for ip in g[group]:
ip_list.append(ip["ip"])
ip_list = list(set(ip_list)) # 去除重复的IP
return ip_list else:
ip_list =[]
return ""
def ip_user(ip):
"""
获取主机连接账号信息
:param ip:
:return:
"""
ip_info = []
for data_ip in data:
for ip_data in data_ip:
for iptest in data_ip[ip_data]:
if ip in iptest["ip"]:
ip_info = [iptest["ip"],iptest["port"],iptest["username"],iptest["password"]]
return ip_info
def host_delete():
try:
host_add = input("请要删除主机IP：").strip()
# 判断是否为IP
if re.match(r"((2[0-4]\d|25[0-5]|[01]?\d\d?)\.){3}(2[0-4]\d|25[0-5]|[01]?\d\d?)$",host_add):
flag = check_ip(host_add)
if flag:
for g in data:
for g_data in g:
for index,host in enumerate(g[g_data]):
if host_add in host["ip"]:
print(g[g_data][index])
del g[g_data][index]
with open(db_path,"wb") as fw:
pickle.dump(data,fw)
log.info("删除主机[%s]成功！"%host_add)
print("删除主机[%s]成功！"%host_add) else:
log.error("删除主机[%s]失败！"%host_add)
print("删除主机[%s]失败！"%host_add)
return add_host()
else:
log.error("你输入的不是IP地址：%s" %host_add)
print("你输入的不是IP地址：%s" %host_add)
except Exception as ex:
log.error("删除主机异常%s"%ex)
print("删除主机异常")
file_handle
#! /usr/bin/env python3.5
# -*- coding:utf-8 -*-
import os,sys,logging,pickle,paramiko
from multiprocessing import Pool
from core import file_handler
BASEDIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.append(BASEDIR)
# 初始化日志格式及对象
logging.basicConfig(level=logging.INFO, filename=os.path.join(BASEDIR,'log/ssh.log'), filemode='a',
format='%(asctime)s %(levelname)s %(message)s', datefmt='%Y/%m/%d %H:%M:%S')
log = logging.getLogger(__name__)
def cmd_func(i,cmd):
ip_info = file_handler.ip_user(i)
if len(ip_info) != 0:
ip = ip_info[0]
port = int(ip_info[1])
username = ip_info[2]
passowrd = ip_info[3]
try:
# 创建SSH对象
ssh = paramiko.SSHClient()
# 允许连接不在know_hosts文件中的主机进行连接
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
# 连接服务器
ssh.connect(hostname=ip,port=port,username=username,password=passowrd)
# 执行命令
resu = []
for excute_cmd in cmd:
stdin, stdout, stderr = ssh.exec_command(excute_cmd)
# 获取结果
result= list(filter(lambda x:x is not None,[stdout.read(),stderr.read()]))[0]
resu.append(result)
# 关闭连接
ssh.close()
log.info("主机：{}，执行命令{}成功！".format(ip,cmd))
for x,result in enumerate(resu): print("主机：{},执行命令：{}，结果如下：\n\n{}".format(ip,cmd[x],result.decode()))
except Exception as e:
print("连接主机{}出错".format(ip))
log.error("连接主机{}出错：{}".format(ip,e))
else:
log.error("没有可用主机可以进行连接")
print("没有可用主机可以进行连接")
def run(arg):
"命令执行方法"
if len(arg) != 2: # 如果arg 没有两个参数
log.info("参数出错，此处需要两个参数{}".format(arg))
print("参数出错，此处需要两个参数{}".format(arg))
else:
# 从元组里拆分出对象列表与指令列表
obj_list ,cmd_list = arg
cmd = " ".join(cmd_list) # 组合命令
cmd = cmd.split(",") # 以逗号分割重组命令
ip_list = file_handler.cmd_handle(obj_list) # 获取所有的IP列表
if len(ip_list) >= 1:
pool = Pool(5)
for i in ip_list:
# cmd_func(i,cmd)
pool.apply_async(cmd_func,args=(i,cmd))
pool.close()
pool.join()
else:
log.info("你当前输入的IP地址不存在，请先增加！")
print("你当前输入的IP地址不存在，请先增加！")
file_handler.add_host()
salt_cmd
#! /usr/bin/env python3.5
# -*- coding:utf-8 -*-
import os,sys,logging,pickle,paramiko
BASEDIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.append(BASEDIR)
from multiprocessing import Pool
from core import file_handler
from conf import setting
# 初始化日志格式及对象
logging.basicConfig(level=logging.INFO, filename=os.path.join(BASEDIR,'log/ssh.log'), filemode='a',
format='%(asctime)s %(levelname)s %(message)s', datefmt='%Y/%m/%d %H:%M:%S')
log = logging.getLogger(__name__)
def file(i,path_x,path_y,obj):
ip_info = file_handler.ip_user(i)
if len(ip_info) != 0:
ip = ip_info[0]
port = int(ip_info[1])
username = ip_info[2]
passowrd = ip_info[3]
try:
transport = paramiko.Transport((ip, port))
transport.connect(username=username, password=passowrd)
except paramiko.ssh_exception.AuthenticationException as e:
# 接收认证错误并返回给结果
log.error("主机:%s,用户名或密码错误,%s"%(ip,e))
print("主机:%s,用户名或密码错误"%ip)
except paramiko.ssh_exception.SSHException as e:
# 接收连接错误并返回给结果
log.error("主机:%s,连接失败:%s"%(ip,e))
print("主机:%s,连接失败"%ip)
else:
sftp = paramiko.SFTPClient.from_transport(transport)
try:
if obj == "get":
if os.path.isdir(os.path.dirname(path_y)):
sftp.get(path_x,path_y)
else:
os.makedirs(os.path.dirname(path_y))
sftp.get(path_x,path_y)
elif obj == "put":
sftp.put(path_x,path_y)
except Exception as e:
log.error("主机:%s,操作失败:%s"%(ip,e))
print("主机:%s,操作失败"%ip)
else:
log.info("主机:%s,文件操作成功！"%ip)
print("主机:%s,文件操作成功！"%ip)
transport.close()
else:
log.error("没有可用主机可以进行连接")
def get(arg):
"命令执行方法"
if len(arg) != 2: # 如果arg 没有两个参数
log.info("参数出错，此处需要两个参数{}".format(arg))
else:
# 从元组里拆分出对象列表与指令列表
obj_list ,file_name = arg
cmd = " ".join(file_name)
remote_path =os.path.join(setting.FILEPATH["remote_path"],cmd) # 组合远程主机目录
ip_list = file_handler.cmd_handle(obj_list) # 获取所有的IP列表
if len(ip_list) >= 1:
pool = Pool(5)
for i in ip_list:
local_path =os.path.join(setting.FILEPATH["loca_path"],i) #组合以IP命令的本地目录
local_path = os.path.join(local_path,cmd)
pool.apply_async(file,args=(i,remote_path,local_path,"get"))
pool.close()
pool.join()
else:
log.info("IP地址为空！可能是输入的IP不合法或没有增加进去") def put(arg):
"命令执行方法"
if len(arg) != 2: # 如果arg 没有两个参数
log.info("参数出错，此处需要两个参数{}".format(arg))
else:
# 从元组里拆分出对象列表与指令列表
obj_list ,file_name = arg
cmd = " ".join(file_name)
local_path =os.path.join(setting.FILEPATH["loca_path"],cmd) # 组合本地目录
if os.path.isfile(local_path):
remote_path =os.path.join(setting.FILEPATH["remote_path"],cmd) # 组合远程主机目录
ip_list = file_handler.cmd_handle(obj_list) # 获取所有的IP列表
if len(ip_list) >= 1:
pool = Pool(5)
for i in ip_list:
pool.apply_async(file,args=(i,local_path,remote_path,"put"))
pool.close()
pool.join()
else:
log.error("IP地址为空！可能是输入的IP不合法或没有增加进去")
else:
log.error("文件%s不存在"%local_path)
salt_file
