包含修改版：
import telnetlib
import time
import pandas as pd
import re
def telnet_login(host_ip, username, password):
tn = telnetlib.Telnet()
try:
tn.open(host_ip, port=23, timeout=5)
tn.read_until(b'Username:', timeout=5)
tn.write(username.encode('ascii') + b'\n')
tn.read_until(b'Password:', timeout=5)
tn.write(password.encode('ascii') + b'\n')
time.sleep(1)
command_result = tn.read_until(b'#', timeout=5)
if b'#' in command_result:
print(f'{host_ip} telnet success')
else:
print(f'{host_ip} telnet failed')
return tn
except:
print(f'{host_ip} telnet failed')
def get_intface_state(tn, host_ip):
command = "show ip interface brief"
cmd = bytes(command, encoding='ascii') # 注意不能写成encodings='ascii'
tn.write(cmd + b'\n')
time.sleep(1)
resultlist = []
while True: # 不要忘了
command_result = tn.read_very_eager().decode('ascii')
# print(command_result)
# 注意不能写成tn.read_very_eager('ascii')
resultlist.append(command_result) # 不能写成resultlist=resultlist.append(command_result)
if re.findall(r'#', command_result.strip()):
break
elif re.findall(r'--More--', command_result.strip()):
tn.write(b" ")
else:
time.sleep(0.05)
continue
resultstr = "\n".join(resultlist)
liststr = resultstr.split('\n')
pd_output = pd.DataFrame()
listtemp = []
for j in liststr:
if len(re.findall(r'Interface\s+.+', j)) > 0:
new_columns = re.split(r'\s+', j.strip())
if len(re.findall(r'.+\d+\.\d+\.\d+\.\d+\s+.+', j)) > 0:
list_find_str = re.split(r'\s+', j.strip())
listtemp.append(list_find_str)
# print(listtemp)
pd_output = pd.DataFrame(listtemp)
pd_output.columns = new_columns
print(pd_output)
pd_output.to_csv(f'{host_ip}.csv')
return pd_output
def get_ospf_config(tn):
try:
tn.write(b"show running-config ospfv2" + b"\n")
time.sleep(1)
result_list = []
command_result = tn.read_very_eager().decode('ascii')
result_list.append(command_result.strip())
result_str = "\n".join(result_list)
result_list = result_str.split('\n')
# print(result_str)
pd_output = pd.DataFrame()
import_list = []
data = []
last_area = []
df_import = pd.DataFrame()
count = 0
# row = None
for i in result_list:
match_area = re.match(r'^\s*area\s+(\d\.\d\.\d\.\d)$', i.strip())
match_interface = re.match(r'^\s*interface\s+(\w+.+)$', i.strip())
match_network = re.match(r'^\s*network\s+(\w+.+)$', i.strip())
match_redistribute = re.match(r'^\s*redistribute\s+(\w+)$', i.strip())
# match_end=re.match(r'^\s*$\s*', i.strip())
if match_area:
data.append({'area': match_area.group(1)})
last_area = match_area.group(1)
count = 0
elif match_interface and count < 2:
if len(data) > 0 and 'interface' not in data[-1].keys():
data[-1]['interface'] = match_interface.group(1)
count = 0
elif len(data) > 0 and 'interface' in data[-1].keys():
entry = {'area': last_area}
entry['interface'] = match_interface.group(1)
data.append(entry)
count = 0
else:
data.append({'interface': match_interface.group(1)})
count = 0
# continue
elif match_network and count < 2:
data[-1]['network'] = match_network.group(1)
count = 0
elif match_redistribute:
findstr = match_redistribute.group(1)
import_list.append(findstr)
count = 0
else:
count += 1
if len(import_list) > 0:
df_import = pd.DataFrame(import_list)
else:
df_import.loc[0, 0] = ''
df_import.columns = ['import']
# print(data)
df_area = pd.DataFrame(data)
if 'area' not in df_area.columns:
df_area['area']='null'
if 'interface' not in df_area.columns:
df_area['interface']='null'
if 'network' not in df_area.columns:
df_area['network']='null'
df = pd.concat([df_area, df_import], axis=1)
# df['host_ip'] = host_ip
print(df)
return df
except:
print('error')
def get_ospf_state(tn):
tn.write(b"show ip ospf neighbor detail" + b'\n')
time.sleep(1)
result_list = []
command_result = tn.read_very_eager().decode('ascii')
result_list.append(command_result.strip())
result_str = "\n".join(result_list)
if re.findall(r'State\s+FULL', result_str):
print('The ospf state is FULL!')
return True
else:
print('The ospf state is not FULL!')
return False
def ping_test(tn, dest_ip, source_ip):
command = f"ping {dest_ip} source {source_ip}"
command = bytes(command, encoding='utf-8')
tn.write(command + b'\n')
time.sleep(10)
ping_result = tn.read_very_eager().decode('ascii')
# 检查ping测试是否成功
print(ping_result)
return "Success rate is 100 percent" in ping_result
def modidy_ospf_config(tn,area,interface,network):
command=[
"enable",
"configure terminal",
"router ospf 1",
f"area {area}",
f"interface {interface}",
f"network {network}",
"exit",
"exit",
"exit",
"exit",
]
for cmd in command:
command = bytes(cmd, encoding='utf-8')
tn.write(command + b'\n')
time.sleep(0.1)
time.sleep(20)
def remove_ospf_config(tn,area):
command=[
"enable",
"configure terminal",
"router ospf 1",
f"no area {area}",
"exit",
"exit",
]
for cmd in command:
command = bytes(cmd, encoding='utf-8')
tn.write(command + b'\n')
time.sleep(0.1)
def compare_ospf_config(tn1, tn2, df1, df2, interface):
network_config = 'point-to-point'
if interface in df1['interface'].values:
if interface in df2['interface'].values:
if df1[df1['interface']==interface]['area'].values[0]==df2[df2['interface']==interface]['area'].values[0]:
if df1[df1['interface']==interface]['network'].values[0]==df2[df2['interface']==interface]['network'].values[0]:
print("two router area configured are ok !")
else:
area_config = df1[df1['interface'] == interface]['area'].values[0]
modidy_ospf_config(tn1,area_config,interface,network_config)
modidy_ospf_config(tn2, area_config, interface, network_config)
else:
area_config=df1[df1['interface']==interface]['area'].values[0]
area_remove=df2[df2['interface']==interface]['area'].values[0]
print('configured areas are not same,trying modify it')
remove_ospf_config(tn2,area_remove)
modidy_ospf_config(tn1, area_config, interface, network_config)
modidy_ospf_config(tn2, area_config, interface, network_config)
else:
area_config = df1[df1['interface'] == interface]['area'].values[0]
print(f'router2 has not configured the interface {interface},trying modify it')
modidy_ospf_config(tn1, area_config, interface, network_config)
modidy_ospf_config(tn2, area_config, interface, network_config)
else:
if interface in df2['interface'].values:
print(f'router1 has not configured the interface {interface},router2 configured,trying modify it')
area_config=df2[df2['interface']==interface]['area'].values[0]
modidy_ospf_config(tn1, area_config, interface, network_config)
modidy_ospf_config(tn2, area_config, interface, network_config)
else:
print(f'two routers have not configured the interface {interface},trying modify them')
area_config='0.0.0.0'
modidy_ospf_config(tn1, area_config, interface, network_config)
modidy_ospf_config(tn2, area_config, interface, network_config)
if 'connected' not in df1['import'].values[0]:
print('router1 has not configuered:import connected')
modify_ospf_import(tn1)
if 'connected' not in df2['import'].values[0]:
print('router2 has not configuered:import connected')
modify_ospf_import(tn2)
get_ospf_state(tn1)
def modify_interface_state(tn, interface):
commands = [
"enable",
"configure terminal",
f"interface {interface}",
"no shutdown",
"exit",
"exit"
]
for cmd in commands:
command = bytes(cmd, encoding='utf-8')
tn.write(command + b'\n')
time.sleep(2) # 等待命令执行
def modify_interface_ip(tn, interface, ip, mask):
commands = [
"enable",
"configure terminal",
f"interface {interface}",
f"ip address {ip}/{mask}",
"exit",
"exit"
]
for cmd in commands:
command = bytes(cmd, encoding='utf-8')
tn.write(command + b'\n')
time.sleep(2)
def modify_ospf_import(tn):
commands = [
"enable",
"configure terminal",
"router ospf 1",
"redistribute connected",
"exit",
"exit"
]
for cmd in commands:
command = bytes(cmd, encoding='utf-8')
tn.write(command + b'\n')
time.sleep(0.1)
if __name__ == '__main__':
hosts_ip = ['192.168.8.1', '192.168.2.2']
interfaces = ['xgei-0/1/1/49', 'loopback1']
planed_ip = ['10.0.0.2', '2.2.2.2', '10.0.0.1', '3.3.3.3']
mask = ['255.255.255.252', '255.255.255.255', '255.255.255.252', '255.255.255.255']
mask_config = ['30', '32', '30', '32']
dest_ip = ['3.3.3.3', '2.2.2.2']
source_ip = ['2.2.2.2', '3.3.3.3']
username = 'zte'
password = 'zte'
df = pd.DataFrame() # 注意不能写成df=DataFrame()
df1 = pd.DataFrame()
df2 = pd.DataFrame()
# tn=telnet_login(host_ip=hosts_ip[0],username=username,password=password)
df_ospf_two = pd.DataFrame()
a = 0
for host_ip in hosts_ip:
tn = telnet_login(host_ip, username, password)
if tn:
df = get_intface_state(tn, host_ip)
for interface in interfaces:
if interface in df['Interface'].values:
if df[df['Interface'] == interface]['Admin'].values[0] == 'down':
print(f'{interface} is admin down, please no shutdown it')
modify_interface_state(tn, interface)
elif df[df['Interface'] == interface]['IP-Address'].values[0] != planed_ip[a] or \
df[df['Interface'] == interface]['Mask'].values[0] != mask[a]:
print(f'{interface} ip address is not {planed_ip[a]} {mask[a]}')
modify_interface_ip(tn, interface, planed_ip[a], mask_config[a])
else:
print(f'{interface} configured is ok')
else:
print(f'{interface} has not configuered ip address')
a = a + 1
tn.close()
else:
print('telnet failed')
tn1 = telnet_login(hosts_ip[0], username, password)
tn2 = telnet_login(hosts_ip[1], username, password)
df1 = get_ospf_config(tn1)
df2 = get_ospf_config(tn2)
interface = interfaces[0]
compare_ospf_config(tn1, tn2, df1, df2, interface)
ping_test(tn1, dest_ip[0], source_ip[0])
标签：area,ip,最终,tn,交换机,command,interface,相关,config From： https://www.cnblogs.com/tjrk/p/18457989