场景：通过Python3自动下发交换机配置
import dns.resolver
from netmiko import ConnectHandler, SSHDetect, Netmiko
from netmiko.exceptions import NetMikoTimeoutException, NetMikoAuthenticationException
import logging
import os
class RunningLogger:
def __init__(self, path,clevel = logging.DEBUG,Flevel = logging.DEBUG):
self.logger = logging.getLogger(path)
self.logger.setLevel(logging.DEBUG)
fmt = logging.Formatter('[%(asctime)s] [%(levelname)s] %(message)s', '%Y-%m-%d %H:%M:%S')
#设置CMD日志
sh = logging.StreamHandler()
sh.setFormatter(fmt)
sh.setLevel(clevel)
#设置文件日志
fh = logging.FileHandler(path)
fh.setFormatter(fmt)
fh.setLevel(Flevel)
self.logger.addHandler(sh)
self.logger.addHandler(fh)
def debug(self,message):
self.logger.debug(message)
def info(self,message):
self.logger.info(message)
def war(self,message):
self.logger.warn(message)
def error(self,message):
self.logger.error(message)
def cri(self,message):
self.logger.critical(message)
running_logger = RunningLogger('Bot_Running.log', logging.ERROR,logging.DEBUG, )
app_redis_db = redis.StrictRedis(host='127.0.0.1',port=6379,db=0,decode_responses=True)
def Is_Valid_IPv4(ipv4):
try:
ipaddress.IPv4Address(ipv4)
return True
except ipaddress.AddressValueError:
return False
def Get_DNS_Record_List(public_dns_ip,domain,query_type='A'):
resolver = dns.resolver.Resolver(configure=False)
resolver.nameservers = [public_dns_ip]
resolver.timeout = 2
resolver.lifetime = 2
try:
answer = resolver.resolve(qname=domain, rdtype=query_type)
dns_record_list = []
for i in answer.response.answer:
for j in i.items:
if Is_Valid_IPv4(j.to_text()):
dns_record_list.append(j.to_text())
return dns_record_list
except Exception as e:
# running_logger.info(f'【INFO】DNS解析失败，{public_dns_ip} {e}！')
return [False,str(e)]
pass
def Execute_CDN_Switch_CMD(name="", mode="", cmd="",cmd_list=[]):
if name == "cdn-1":
ip = "103.1.1.1"
elif name == "cdn-2":
ip = "103.1.1.2"
elif name == "cdn-3":
ip = "103.1.1.3"
try:
device = {
'device_type': 'hp_comware',
'ip': ip,
'username': 'blueduck',
'password': 'blueduck',
'port': 22,
# 'verbose': True,
'session_log': 'session.log',
'timeout' : 180,
'secret': ''
}
i = 0
cmd_result = ''
loop_delay = 1
timeout = 100
loops = timeout / loop_delay
end_pattern1 = '>'
end_pattern2 = ']'
with ConnectHandler(**device) as conn:
conn.enable()
# config_cmds = ['ip route-static 114.1.1.1 32 100.65.0.2','ip route-static 114.2.2.2 32 100.65.0.2']
config_cmds = cmd_list
conn.send_config_set(config_commands=config_cmds)
# conn.write_channel('{}{}'.format(f'{cmd}',conn.RETURN))
if mode == "read":
while i <= loops:
output = conn.read_channel()
if output:
cmd_result = cmd_result + output
# print("output____:", output)
time.sleep(loop_delay)
i = i + 1
if i > 1 and (end_pattern1 in output or end_pattern2 in output):
break
if mode == "read":
print('执行结果为：{}'.format(cmd_result))
IPv4_Pattern = r'\b(?:\d{1,3}\.){3}\d{1,3}\b'
IPv4_Addr_List = re.findall(IPv4_Pattern, cmd_result)
IPv4_Addr_List = list(set(IPv4_Addr_List))
if IPv4_Addr_List:
print(IPv4_Addr_List)
conn.disconnect()
except (EOFError, NetMikoTimeoutException):
running_logger.error("NetMiko: Can not connect to H3C Switch")
except (EOFError, NetMikoAuthenticationException):
running_logger.error("NetMiko: Wrong username or password")
except Exception as e:
running_logger.error("NetMiko: Wrong enable password")
# Execute_CDN_Switch_CMD(name="cdn-1", mode="read", cmd="dis ip routing-table | in 100.65.0.2")
def Deplopy_Oversea_Task():
if file_name == "SecOpsBot":
OverseaLine_IP_Domains = "OverseaLine_IP_Domains"
OverseaLine_VM_Destionation_IP_Relationship = "OverseaLine_VM_Destionation_IP_Relationship"
else:
OverseaLine_IP_Domains = "OverseaLine_IP_Domains_Test"
OverseaLine_VM_Destionation_IP_Relationship = "OverseaLine_VM_Destionation_IP_Relationship_Test"
oversea_ip_domains_in_redis = app_redis_db.get(OverseaLine_IP_Domains)
oversea_vm_destionation_ip_in_redis = app_redis_db.get(OverseaLine_VM_Destionation_IP_Relationship)
if oversea_vm_destionation_ip_in_redis:
oversea_vm_destionation_ip_dict = json.loads(oversea_vm_destionation_ip_in_redis)
else:
oversea_vm_destionation_ip_dict = {}
# 如果有新增业务IP和域名数据就开始往下执行
if oversea_ip_domains_in_redis:
oversea_ip_domains_dict = json.loads(oversea_ip_domains_in_redis)
switch_config_cmd = []
for vm_ip in oversea_ip_domains_dict.keys():
# 判断vm ip有没有存在的关联数据，有关联数据，需要检查是否有更新的目的IP地址需要添加
if oversea_vm_destionation_ip_dict.get(vm_ip):
dns_record_list = []
new_DstIP_list = []
for domain in oversea_ip_domains_dict.get(vm_ip):
dns_record_list += Get_DNS_Record_List(public_dns_ip="172.16.16.16",domain=domain,query_type='A')
dns_record_list = list(set(dns_record_list))
for DstIP in dns_record_list:
if DstIP not in oversea_vm_destionation_ip_dict.get(vm_ip):
# 如果解析出来的IP不在已有数据里，需要更新
new_DstIP_list.append(DstIP)
new_DstIP_list = list(set(new_DstIP_list))
for item in new_DstIP_list:
switch_config_cmd.append(f'ip route-static {item} 32 {vm_ip}')
running_logger.info(f"有关联数据,新增目的IP, ip route-static {item} 32 {vm_ip}")
# 将增量数据生成交换机命令，然后统一执行
all_dns_record = list(set(new_DstIP_list + oversea_vm_destionation_ip_dict[vm_ip]))
oversea_vm_destionation_ip_dict[vm_ip] = all_dns_record
# 保存至redis中
app_redis_db.set(OverseaLine_VM_Destionation_IP_Relationship, json.dumps(oversea_vm_destionation_ip_dict))
else:
# redis里没有vm和转发目的IP的关联数据，需要初始化数据，然后下发静态路由至交换机
#动态解析域名获取目的IP，并保存起来，保存格式为
# {
# '100.65.0.1': [1.1.1.1,2.2.2.2]
#}
dns_record_list = []
for domain in oversea_ip_domains_dict.get(vm_ip):
dns_record_list += Get_DNS_Record_List(public_dns_ip="172.16.16.16",domain=domain,query_type='A')
dns_record_list = list(set(dns_record_list))
oversea_vm_destionation_ip_dict[vm_ip] = dns_record_list
# 将增量数据生成交换机命令，然后统一执行
for DstIP in dns_record_list:
switch_config_cmd.append(f'ip route-static {DstIP} 32 {vm_ip}')
running_logger.info(f"初始化，新增目的IP,ip route-static {DstIP} 32 {vm_ip} ")
# 保存至redis中
app_redis_db.set(OverseaLine_VM_Destionation_IP_Relationship, json.dumps(oversea_vm_destionation_ip_dict))
switch_config_cmd = list(set(switch_config_cmd))
if switch_config_cmd:
Execute_CDN_Switch_CMD(name="cdn-1", cmd_list=switch_config_cmd)
# Deplopy_Oversea_Task()
