สำหรับใครที่ใช้งาน Secure CRT อยู่เป็นประจำและมีงานที่ต้องดึงข้อมูลโดยใช้ command line อยู่เป็นประจำหรือต้องดึงข้อมูลปริมาณมาก การใช้งาน Script เพื่อช่วยงานจะเป็นเครื่องมือที่ช่วยทุ่นแรงเราอย่างมาก การใช้งานก็ไม่ได้ยากเท่าใดนัก ซึ่งในบทความนี้ผมจะยกตัวอย่างที่ผมคิดว่าใช้งานบ่อยๆ ครับ
Script ที่ผมใช้รัน จะ simulate ในโน๊ตบุ๊คโดยใช้ EVE-NG ซิมอุปกรณ์ Cisco และ Nokia Tology ก็จะตามรูปด้านล่างนี้ครับ
บทที่ 1 Telnet และ SSH2
เอาโค้ดไปก่อนเดี๋ยวมาเล่าให้ฟัง
1.1 Nokia_telnet.py
nokia_ip = "192.168.158.2"
nokia_username = "admin"
nokia_password = "admin"
objtab = crt.Session.ConnectInTab("/telnet 192.168.158.2",True , True)
objtab.Caption = "Nokia_Telnet"
if objtab.Session.Connected:
objtab.Screen.Synchronous = True
objtab.Screen.WaitForString("Login")
objtab.Screen.Send(nokia_username + "\n")
objtab.Screen.WaitForString("Password:")
objtab.Screen.Send(nokia_password + "\n")
objtab.Screen.WaitForString("#")
# do something here
1.2 Nokia_ssh2.py
nokia_ip = "192.168.158.2"
username = "admin"
password = "admin"
word = "/SSH2 /ACCEPTHOSTKEYS /PASSWORD %s %s@%s "%(username, password, nokia_ip)
objtab = crt.Session.ConnectInTab(word , True, True)
objtab.Caption = "Nokia SSH2"
if objtab.Session.Connected:
objtab.Screen.Synchronous = True
# do something here
1.3 Cisco_telnet.vbs
cisco_ip = "192.168.158.3"
cisco_username = "cisco"
cisco_password = "cisco"
Set objtab = crt.Session.ConnectInTab("/telnet 192.168.158.3",True , True)
objtab.Caption = "Cisco_Telnet"
If objtab.Session.Connected Then
objtab.Screen.Synchronous = True
objtab.Screen.WaitForString("Username:")
objtab.Screen.Send(cisco_username & vbCr)
objtab.Screen.WaitForString("Password:")
objtab.Screen.Send(cisco_password & vbCr)
' do something here
End If
1.4 Cisco_ssh2.vbs
cisco_ip = "192.168.158.3"
username = "cisco"
password = "cisco"
word = "/SSH2 /ACCEPTHOSTKEYS /PASSWORD " & username & " " + password & "@" & cisco_ip
Set objtab = crt.Session.ConnectInTab( word, True, True)
objtab.Caption = "Cisco_Telnet"
If objtab.Session.Connected Then
objtab.Screen.Synchronous = True
' do something here
End If
บทที่ 2 log
เอาโค้ดไปก่อนเดี๋ยวมาเล่าให้ฟัง
2.1 Nokia_log.py
#-- to avoid mission data
crt.Screen.Synchronous = True
#-- release log
crt.Session.Log(False)
#-- start log session, folder "tun" auto create
crt.Session.LogFileName = "Log_Nokia/2.1 nokia_log result.txt"
#--Log(True,True) = append Log(True,False) = overwrite
crt.Session.Log(True,True)
#-- send command line and wait for string
crt.Screen.Send("environment no more\n")
crt.Screen.WaitForString("#")
crt.Screen.Send("show port description\n")
crt.Screen.WaitForString("#")
crt.Screen.Send("show router interface\n")
crt.Screen.WaitForString("#")
#-- release log
crt.Session.Log(False)
#-- display message diaglog
crt.Dialog.MessageBox("Finish", "Log")
2.2 Cisco_log.vbs
'-- Synchronous with the screen
crt.Screen.Synchronous = True
'-- release log session
crt.Session.Log False
'-- start log session
crt.Session.LogFileName = "Log_Cisco/2.2 Cisco_log result.txt"
'-- Log True, True = append, Log True, False = overwrite
crt.Session.Log True, False
'-- send command and wait for string
crt.Screen.Send("terminal length 0" & vbCr)
crt.Screen.WaitForString("#")
crt.Screen.Send("show interfaces description" & vbCr)
crt.Screen.WaitForString("#")
crt.Screen.Send("show ipv4 interface brief" & vbCr)
crt.Screen.WaitForString("#")
'-- release log session
crt.session.Log False
'-- show message diaglog
crt.Dialog.MessageBox("Finish")
บทที่ 3 Chapter 3 Login + Log and Improvement
Login แล้วก็เก็บ log. มีการปรับปรังโค้ดบ้างเช่นเช็ค directory. แสดง dialog box หลัง run เสร็จ
เพียงแค่นี้เราก็สามารถทำงานที่เยอะๆ หรือซ้ำ ได้แล้ว
3.1 Nokia_login_log_improve.py
import os
import time
#-- device parameter --
nokia_ip = "192.168.158.2"
username = "admin"
password = "admin"
#-- get script path located example = C:\Users\pick_\Desktop\script
working_dir = os.path.dirname(os.path.realpath(__file__))
#-- login by SSH2
word = "/SSH2 /ACCEPTHOSTKEYS /PASSWORD %s %s@%s "%(username, password, nokia_ip)
objtab = crt.Session.ConnectInTab(word , True, True)
objtab.Caption = "Nokia_Log"
if objtab.Session.Connected:
#-- Screen Sync to avoid missing data and skip first # when login finish
objtab.Screen.Synchronous = True
objtab.Screen.WaitForString("#")
#-- set log_fullpath
new_folder = "Log_Nokia"
logname = "3.1 Nokia_log result.txt"
log_fullpath = os.path.join(working_dir, new_folder, logname)
#-- start log, folder "Log_Nokia" auto create if not exist, overwrite to file
objtab.Session.Log(False)
objtab.Session.LogFileName = log_fullpath
objtab.Session.Log(True, False)
#-- send command and wait for string
objtab.Screen.Send("environment no more\n")
objtab.Screen.WaitForString("#")
objtab.Screen.Send("show port description\n")
objtab.Screen.WaitForString("#")
objtab.Screen.Send("show router interface\n")
objtab.Screen.WaitForString("#")
objtab.Screen.Send("\n")
objtab.Screen.WaitForString("#")
objtab.Screen.Send("logout\n")
#-- sleep 2 sec and release log
time.sleep(2) # (sec)
objtab.Session.Log(False)
#-- show message diaglog
crt.Dialog.MessageBox("Finish")
3.2 Cisco_login_log_improve.vbs
'-- device parameter --
cisco_ip = "192.168.158.3"
username = "cisco"
password = "cisco"
'-- get script path located example = C:\Users\pick_\Desktop\script
Set FSO = CreateObject("Scripting.FileSystemObject")
working_dir = FSO.GetAbsolutePathName(".")
'-- login by SSH2
word = "/SSH2 /ACCEPTHOSTKEYS /PASSWORD " & username & " " + password & "@" & cisco_ip
Set objtab = crt.Session.ConnectInTab(word, True, True)
objtab.Caption = "Cisco_Log"
If objtab.Session.Connected Then
'-- screen sync and skip first # when login finish
objtab.Screen.Synchronous = True
objtab.Screen.WaitForString("#")
'-- set log_fullpath
new_folder = "Log_Cisco"
logname = "3.2 Cisco_log result.txt"
log_fullpath = FSO.BuildPath(log_path, new_folder, logname)
'-- start log, folder "Log_Cisco" auto create if not exist, overwrite
objtab.Session.Log False
objtab.Session.LogFileName = log_fullpath
objtab.Session.Log True, False
'-- send command and wait for string
objtab.Screen.Send("terminal length 0" & vbCr)
objtab.Screen.WaitForString("#")
objtab.Screen.Send("show interfaces description" & vbCr)
objtab.Screen.WaitForString("#")
objtab.Screen.Send("show ipv4 interface brief" & vbCr)
objtab.Screen.WaitForString("#")
objtab.Screen.Send(vbCr)
objtab.Screen.WaitForString("#")
objtab.Screen.Send("exit" & vbCr)
'-- sleep 2 sec and release log
crt.Sleep 2000 ' (ms)
objtab.session.Log False
'-- show message diaglog
crt.Dialog.MessageBox("Finish")
End If
บทที่ 4 Readstring
Login แล้วก็เก็บ log. มีการปรับปรังโค้ดบ้างเช่นเช็ค directory. แสดง dialog box หลัง run เสร็จ
เพียงแค่นี้เราก็สามารถทำงานที่เยอะๆ หรือซ้ำ ได้แล้ว
ในตัวอย่างนี้จะใช้ command show router arp โดย python ผมจะใช้ regular expression ในการช่วยตัด text
ส่วน vbs ผมจะตัดตาม character
4.1 Nokia_read_string.py
import re
nokia_ip = "192.168.158.2"
username = "admin"
password = "admin"
#-- login by SSH2
word = "/SSH2 /ACCEPTHOSTKEYS /PASSWORD %s %s@%s "%(username, password, nokia_ip)
objtab = crt.Session.ConnectInTab(word , True, True)
objtab.Caption = "Nokia_Ping"
if objtab.Session.Connected:
#-- Screen Sync and skip first # when login finish
objtab.Screen.Synchronous = True
objtab.Screen.WaitForString("#")
#-- send command and wait for string
objtab.Screen.Send("environment no more\n")
objtab.Screen.WaitForString("#")
#-- get output from show router arp
objtab.Screen.Send("show router arp\n")
data = objtab.Screen.ReadString("#")
#-- read data line by line then extract ip store to arp_list
arp_list = []
re_arp = re.compile(r"(\d+\.\d+\.\d+\.\d+).*Dyn")
for line in data.split("\n"):
if re_arp.search(line):
arp_ip = re_arp.search(line).group(1)
arp_list.append(arp_ip)
#-- for this exmaple arp_list = ["192.168.10.2", "192.168.20.2"]
#-- ping
for arp_ip in arp_list:
objtab.Screen.Send("ping %s rapid\n"%(arp_ip))
objtab.Screen.WaitForString("#")
#-- log out
objtab.Screen.Send("\n")
objtab.Screen.WaitForString("#")
objtab.Screen.Send("logout\n")
#-- show message diaglog
crt.Dialog.MessageBox("Ping Finish","Result")
4.2 Cisco_read_string.vbs
cisco_ip = "192.168.158.3"
username = "cisco"
password = "cisco"
'-- login by SSH2
word = "/SSH2 /ACCEPTHOSTKEYS /PASSWORD " & username & " " + password & "@" & cisco_ip
Set objtab = crt.Session.ConnectInTab( word, True, True)
objtab.Caption = "Cisco_Ping"
If objtab.Session.Connected Then
'-- screen sync and skip first # when login finish ??
objtab.Screen.Synchronous = True
objtab.Screen.WaitForString("#")
'-- send command and wait for string
objtab.Screen.Send("terminal length 0" & vbCr)
objtab.Screen.WaitForString("#")
'-- get output from show router arp
crt.Screen.Send("show arp" & vbCr)
data = crt.Screen.ReadString("#")
'-- show output to check data captured
crt.Dialog.MessageBox(data)
'-- split line by line and get arp_ip to one line string
lines = Split(data, vbCrLf)
arp_line = ""
For each aline in lines
If InStr(aline,"Dynamic") and InStr(aline,"GigabitEthernet") Then
arp_ip = Split(aline, " ")(0)
arp_line = arp_line & " " & arp_ip
End If
Next
'-- result arp_line = " 192.168.10.2 192.168.20.2"
'-- convert one line string to array then ping
arp_list = Split(Trim(arp_line), " ")
For each arp_ip in arp_list
objtab.Screen.Send("ping " & arp_ip & vbCr)
objtab.Screen.WaitForString("#")
Next
'-- log out
objtab.Screen.Send(vbCr)
objtab.Screen.WaitForString("#")
objtab.Screen.Send("exit" & vbCr)
'-- show message diaglog
crt.Dialog.MessageBox("Ping Finish")
End If
บทที่ 5 python read/write .csv
1) py script จะเปิดไฟล์ 5.1 ip_list.csv เพื่ออ่าน ip , username, password
2) Telnet และใช้ WaitForStrings แบบมีเงื่อนไขเพื่อเช็คว่าเป็น Login แบบ Nokia หรือ Cisco
3) เก็บข้อมูล port, status และ description
4) save as .csv
5.1 ip_list.csv
5.2 Python_read_write_csv.py
import csv
import re
import os
import time
working_dir = os.path.dirname(os.path.realpath(__file__))
csv_file = "5.1 ip_list.csv"
re_ip = re.compile("(\d+\.\d+\.\d+\.\d+)")
#-- read csv file, return as dictionnary
def read_csv():
site_info = {}
csv_fullpath = os.path.join(working_dir, csv_file)
with open(csv_fullpath, 'r') as f:
reader = csv.reader(f, delimiter=',')
for row in reader:
ip = row[0] #-- column A
username = row[1] #-- column B
password = row[2] #-- column B
#-- check is ip format
if re_ip.search(ip):
site_info[ip] = {"username": username, "password": password}
return site_info
def login(ip, username, password):
objtab = crt.Session.ConnectInTab("/telnet %s"%(ip),True , True)
objtab.Caption = "py_csv"
hostname = ""
brand = ""
if objtab.Session.Connected:
objtab.Screen.Synchronous = True
#-- WaitForStrings multiple condition and timeout is 15 seconds
result = objtab.Screen.WaitForStrings(["Login:","Username:"], 15)
time.sleep(1)
if result == 1:
brand = "Nokia"
objtab.Screen.Send(username + "\n")
objtab.Screen.WaitForString("Password:")
objtab.Screen.Send(password + "\n")
data = objtab.Screen.ReadString("#")
elif result == 2:
brand = "Cisco"
objtab.Screen.Send(username + "\n")
objtab.Screen.WaitForString("Password:")
objtab.Screen.Send(password + "\n")
data = objtab.Screen.ReadString("#")
elif result == 0:
#-- timeout reached
crt.Dialog.MessageBox("Time out")
return brand, objtab
#-- write port status to csv
def write_csv(info):
save_csv = os.path.join(working_dir, "5.2 result interface status.csv")
with open(save_csv, "wb") as f:
acsv = csv.writer(f,delimiter=',')
field = ["IP", "Interface", "Admin Status", "Port State"]
acsv.writerow(field)
for ip in info:
for [port, admin_state, port_state] in info[ip]:
rowdata = [ip, port, admin_state, port_state ]
acsv.writerow(rowdata)
def main():
#-- read ip, username, password from csv file
site_info = read_csv()
info = {}
for ip in site_info:
info[ip] = []
username = site_info[ip]["username"]
password = site_info[ip]["password"]
#-- login to each node
brand, objtab = login(ip, username, password)
#-- check brand
if brand == "Nokia":
objtab.Screen.Send("environment no more \n")
objtab.Screen.WaitForString("#")
#-- send command and capture output
objtab.Screen.Send("show port\n")
data = objtab.Screen.ReadString("#")
#-- extract output line by line and extract only port, admin state and port state
re_nokia_port = re.compile(r"(\d+\/\d+\/\d+)\s+(Up|Down).*(Up|Down)\s+")
for aline in data.split("\n"):
if re_nokia_port.search(aline):
#-- nokia port is the same as date in excel. example 1/1/1 -> 1/1/2001(excel)
#-- add something to avoid auto convert to date when open in excel
port = re_nokia_port.search(aline).group(1)
port = "'" + port
admin_state = re_nokia_port.search(aline).group(2)
port_state = re_nokia_port.search(aline).group(3)
#-- record port status. use list instead of dictionary because dict not ordered
port_list = [port, admin_state, port_state]
info[ip].append(port_list)
#-- logout and sleep for 1 sec
objtab.Screen.Send("logout\n")
time.sleep(1)
elif brand == "Cisco":
objtab.Screen.Send("terminal length 0\n")
objtab.Screen.WaitForString("#")
#-- send command and capture output
objtab.Screen.Send("show ipv4 interface brief\n")
data = objtab.Screen.ReadString("#")
#-- extract output
re_cisco_port = re.compile(r"(GigabitEthernet[\d|\/]+)\s+\S+\s+(\w+)\s+(\w+)")
for aline in data.split("\n"):
if re_cisco_port.search(aline):
port = re_cisco_port.search(aline).group(1)
admin_state = re_cisco_port.search(aline).group(2)
port_state = re_cisco_port.search(aline).group(3)
#-- record port status. use list because ordered
port_list = [port, admin_state, port_state]
info[ip].append(port_list)
#-- logout and sleep for 1 sec
objtab.Screen.Send("exit\n")
time.sleep(1)
#-- write to csv
write_csv(info)
crt.Dialog.MessageBox("Save as 5.2 result interface interface.csv","Finish")
main()
บทที่ 6 vbs read/write .xlsx
1) vbs จะเปิดไฟล์(ไฟล์จะต้องปิดอยู่) เพื่ออ่าน ip , username, password ใน sheet ip_list
2) Telnet และใช้ WaitForStrings แบบมีเงื่อนไขเพื่อเช็คว่าเป็น Login แบบ Nokia หรือ Cisco
3) เก็บข้อมูล port, status และ description
4) เขียนข้อมูลลงใน sheet "port_status" ที่สร้างไว้แล้ว
5) save ทับไฟล์เดิม
6.1 result interface status.xlsx
สร้างไฟล์เอาไว้แล้ว มี 2 sheet
1. sheet ip เอาไว้เก็บข้อมูล ip username password
2. sheet interface เอาเขียนข้อมูลที่อ่านมาจาก router
6.2 VBS_read_write_xlsx.vbs
Set FSO = CreateObject("Scripting.FileSystemObject")
working_dir = FSO.GetAbsolutePathName(".")
excelfile = "6.1 result interface status.xlsx"
excel_fullpath = FSO.BuildPath(working_dir, excelfile)
Set objExcel = CreateObject("Excel.Application")
Set wb = objExcel.Workbooks.Open(excel_fullpath)
Set ws_ip = wb.Sheets("ip")
Set ws_intf = wb.Sheets("interface")
objExcel.Application.visible = true
ws_intf.Activate
rows_ip = ws_ip.usedrange.rows.count
'-- i represent as current row sheet rows_ip
'-- n represent as current row sheet interface
n = 1
for i = 2 to rows_ip
brand = ""
'-- cells(row, column) start from 1
ip = ws_ip.cells(i, 1).Value
username = ws_ip.cells(i, 2).Value
password = ws_ip.cells(i, 3).Value
Set objtab = crt.session.ConnectInTab("/telnet " & ip)
objtab.Screen.Synchronous = True
result = objtab.Screen.WaitForStrings("Login:", "Username:", 15)
crt.Sleep 1000
If result = 1 Then
brand = "Nokia"
objtab.Screen.Send(username & vbCr)
objtab.Screen.WaitForString("Password:")
objtab.Screen.Send(password & vbCr)
objtab.Screen.WaitForString("#")
objtab.Screen.Send("environment no more" & vbCr)
objtab.Screen.WaitForString("#")
objtab.Screen.Send("show port" & vbCr)
data = objtab.Screen.ReadString("#")
lines = Split(data, vbCrLf)
For each aline in lines
If InStr(aline," xcme ") Then
' 1 2 3 4 5 6 7
'12345678901234567890123456789012345678901234567890123456789012345678901234567890
'1/1/1 Up Yes Up 8704 8704 - netw null xcme GIGE-LX 10KM"
port = Trim(Mid(aline,1, 14))
port = "'" & port
admin_state = Trim(Mid(aline,15, 6))
port_state = Trim(Mid(aline,26, 8))
n = n + 1
ws_intf.Cells(n ,1).Value = ip
ws_intf.Cells(n ,2).Value = port
ws_intf.Cells(n ,3).Value = admin_state
ws_intf.Cells(n ,4).Value = port_state
End If
Next
objtab.Screen.Send("logout" & vbCr)
crt.Sleep 1000
ElseIf result = 2 Then
brand = "Cisco"
objtab.Screen.Send(username & vbCr)
objtab.Screen.WaitForString("Password:")
objtab.Screen.Send(password & vbCr)
objtab.Screen.WaitForString("#")
objtab.Screen.Send("terminal length 0" & vbCr)
objtab.Screen.WaitForString("#")
objtab.Screen.Send("show ipv4 interface brief" & vbCr)
data = objtab.Screen.ReadString("#")
'-- split to array by new line and extract focus txt
lines = Split(data, vbCrLf)
For each aline in lines
If InStr(aline,"GigabitEthernet") Then
' 1 2 3 4 5 6 7
'1234567890123456789012345678901234567890123456789012345678901234567890123
'GigabitEthernet0/0/0/1 192.168.10.2 Up Up"
port = Trim(Mid(aline,1, 30))
admin_state = Trim(Mid(aline,48, 20))
port_state = Trim(Mid(aline,70, 10))
n = n + 1
ws_intf.Cells(n ,1).Value = ip
ws_intf.Cells(n ,2).Value = port
ws_intf.Cells(n ,3).Value = admin_state
ws_intf.Cells(n ,4).Value = port_state
End If
Next
objtab.Screen.Send("exit" & vbCr)
crt.Sleep 1000
ElseIf result = 0 Then
crt.Dialog.MessageBox("Time out")
End If
Next
wb.Save
wb.Close
objExcel.Application.Quit
บทที่ 7 python subprocess to write data to excel
1) vbs จะเปิดไฟล์(ไฟล์จะต้องปิดอยู่) เพื่ออ่าน ip , username, password ใน sheet ip_list
2) Telnet และใช้ WaitForStrings แบบมีเงื่อนไขเพื่อเช็คว่าเป็น Login แบบ Nokia หรือ Cisco
3) เก็บข้อมูล port, status และ description
4) เขียนข้อมูลลงใน sheet "port_status" ที่สร้างไว้แล้ว
5) save ทับไฟล์เดิม
7.1_write_to_excel.exe
import os
import re
import openpyxl
import argparse
from openpyxl.styles import PatternFill
#-- extract argument from pyscript
parser = argparse.ArgumentParser()
parser.add_argument('--action', type=str, required=True)
parser.add_argument('--excel', type=str, required=True)
parser.add_argument('--log', type=str, required=True)
args = parser.parse_args()
excelfile = args.excel
log_dir = args.log
action = args.action
print(f"action={action} excelfile={excelfile} log_dir={log_dir} ")
def read_log():
info = {}
re_nokia_hostname = re.compile(r":(.*)# show port ")
re_cisco_hostname = re.compile(r":(.*)#show ipv4 interface brief")
re_nokia_port = re.compile(r"(\d+\/\d+\/\d+)\s+(Up|Down).*(Up|Down)\s+")
re_cisco_port = re.compile(r"(GigabitEthernet[\d+|\/]+)\s+\S+\s+(\S+)\s+(\S+)")
re_ip_txt = re.compile(r"(\d+\.\d+\.\d+\.\d+).*\.txt$")
#-- list all file in log_dir
for afile in os.listdir(log_dir):
#-- read text file line by line
if re_ip_txt.search(afile):
print(f"reading file = {afile}")
host_ip = re_ip_txt.search(afile).group(1)
info[host_ip] = {"hostname":"", "port_status":[]}
fullfile = os.path.join(log_dir, afile)
with open(fullfile, 'r', encoding='UTF-8') as text_file:
hostname = ""
for aline in text_file:
#-- get hostname
if re_nokia_hostname.search(aline):
hostname = re_nokia_hostname.search(aline).group(1)
info[host_ip]["hostname"] = hostname
elif re_cisco_hostname.search(aline):
hostname = re_cisco_hostname.search(aline).group(1)
info[host_ip]["hostname"] = hostname
# -- extract data from string
elif hostname and re_nokia_port.search(aline):
port = re_nokia_port.search(aline).group(1)
# -- avoid excel convert to datetime format
port = f"'{port}"
admin_state = re_nokia_port.search(aline).group(2)
port_state = re_nokia_port.search(aline).group(3)
info[host_ip]["port_status"].append([port, admin_state, port_state])
elif hostname and re_cisco_port.search(aline):
port = re_cisco_port.search(aline).group(1)
admin_state = re_cisco_port.search(aline).group(2)
port_state = re_cisco_port.search(aline).group(3)
info[host_ip]["port_status"].append([port, admin_state, port_state])
elif f"{hostname}#" in aline:
hostname = ""
print("read_log finish\n info = ")
print(info)
return info
def write_excel(info):
if action == "create":
# -- create new excel, new sheet and delete default sheet
wb = openpyxl.Workbook()
del wb['Sheet']
sheetname = "interface"
ws = wb.create_sheet(sheetname, 0)
# -- first row, cell color = blue ---------------
field = ["Hostname","Host_IP", "Port", "Admin_state", "Port_state"]
iblue = "00B0F0"
for h in field:
index = field.index(h)
ws.cell(row=1, column=index + 1).fill = PatternFill("solid", fgColor=iblue)
ws.cell(row=1, column=index + 1).value = h
# -- write data to excel ---------------
i = 1
for host_ip in info:
hostname = info[host_ip]["hostname"]
for [port, admin_state, port_state] in info[host_ip]["port_status"]:
i = i + 1
ws.cell(row=i, column=1).value = hostname
ws.cell(row=i, column=2).value = host_ip
ws.cell(row=i, column=3).value = port
ws.cell(row=i, column=4).value = admin_state
ws.cell(row=i, column=5).value = port_state
# -- save and close
print(f"Save as {excelfile}")
wb.save(excelfile)
wb.close()
def main():
info = read_log()
print("read log finish")
write_excel(info)
print("finish all")
main()
7.2 python call_subprocess
import os
import time
working_dir = os.path.dirname(os.path.realpath(__file__))
log_dir = os.path.join(working_dir,"Log_7.1_subprocess")
def login(ip, username, password):
objtab = crt.Session.ConnectInTab("/telnet %s"%(ip),True , True)
objtab.Caption = "7.1 suprocess"
brand = ""
if objtab.Session.Connected:
objtab.Screen.Synchronous = True
#-- WaitForStrings multiple condition and timeout is 15 seconds
result = objtab.Screen.WaitForStrings(["Login:","Username:"], 15)
time.sleep(1)
if result == 1:
brand = "Nokia"
objtab.Screen.Send(username + "\n")
objtab.Screen.WaitForString("Password:")
objtab.Screen.Send(password + "\n")
data = objtab.Screen.ReadString("#")
elif result == 2:
brand = "Cisco"
objtab.Screen.Send(username + "\n")
objtab.Screen.WaitForString("Password:")
objtab.Screen.Send(password + "\n")
data = objtab.Screen.ReadString("#")
elif result == 0:
#-- timeout reached
crt.Dialog.MessageBox("Time out")
return brand, objtab
def main():
#-- site info
site_info = {}
site_info["192.168.158.2"] = {"username":"admin", "password":"admin"}
site_info["192.168.158.3"] = {"username":"cisco", "password":"cisco"}
info = {}
for ip in site_info:
info[ip] = []
username = site_info[ip]["username"]
password = site_info[ip]["password"]
#-- login to each node
brand, objtab = login(ip, username, password)
#-- start log, folder auto create if not exist, overwrite to file
logname = "7.1 Log_%s_interface_status.txt"%(ip)
log_fullpath = os.path.join(log_dir, logname)
objtab.Session.Log(False)
objtab.Session.LogFileName = log_fullpath
objtab.Session.Log(True, False)
#-- check brand
if brand == "Nokia":
objtab.Screen.Send("environment no more \n")
objtab.Screen.WaitForString("#")
#-- send command and capture output
objtab.Screen.Send("show port\n")
objtab.Screen.WaitForString("#")
objtab.Screen.Send("\n")
objtab.Screen.WaitForString("#")
#-- logout and sleep for 1 sec
objtab.Screen.Send("logout\n")
time.sleep(1)
elif brand == "Cisco":
objtab.Screen.Send("terminal length 0\n")
objtab.Screen.WaitForString("#")
#-- send command and capture output
objtab.Screen.Send("show ipv4 interface brief\n")
objtab.Screen.WaitForString("#")
#-- send enter
objtab.Screen.Send("\n")
objtab.Screen.WaitForString("#")
#-- logout and sleep for 1 sec
objtab.Screen.Send("exit\n")
time.sleep(1)
#-- release log sleep for 1 sec before start subprocess
objtab.Session.Log(False)
time.sleep(1)
#-- convert path to string
excelfile = os.path.join(log_dir, "7.1_result_subprocess.xlsx")
excel = os.path.abspath(excelfile)
log = os.path.abspath(log_dir)
#-- call subprocess
isubprocess = "7.1_write_to_excel.exe"
subprocess.check_call([isubprocess, "--action create", "--excel", excel , "--log", log])
crt.Dialog.MessageBox("subprocess called","Finish")
main()
บทที่ 8 paramiko
1)
2)
3)
4)
5)
import re
import openpyxl
from time import sleep
from paramiko import SSHClient, AutoAddPolicy
from openpyxl.styles import PatternFill
def my_readstring(command_line, session, waitstring, wait_counter=2400):
output = ""
sleep(0.3)
session.send(f"{command_line}\n")
# -- get output from session until found waitstring
i = 0
while i < wait_counter:
i += 1
sleep(0.5)
if session.recv_ready():
data = session.recv(65535).decode('utf-8')
output += data
if waitstring and waitstring in data:
break
# -- reached counter -> log to filed.txt
if i == wait_counter:
with open('log_failed.txt', 'a') as ff:
ff.write(command_line + '\n')
ff.write(output)
return output.replace('\b ', '')
def write_excel(excel_info):
wb = openpyxl.Workbook()
del wb['Sheet']
sheetname = "interface"
ws = wb.create_sheet(sheetname, 0)
# -- first row, cell color = blue ---------------
column = ["A", "B", "C", "D", "E"]
field = ["Hostname", "Host_IP", "Port", "Admin_state", "Port_state"]
iblue = "00B0F0"
for h in field:
index = field.index(h)
ws.cell(row=1, column=index + 1).fill = PatternFill("solid", fgColor=iblue)
ws.cell(row=1, column=index + 1).value = h
ws.column_dimensions[column[index]].width = 20 # set column width = 20
# -- write cell value
i = 1 # row
for excel_row in excel_info:
i = i + 1
j = 0 # column
for cell_value in excel_row:
j = j + 1
ws.cell(row=i, column=j).value = cell_value
# -- save as .xlsx and close
excelfile = "8.1 result interface by paramiko.xlsx"
print(f"Save as {excelfile}")
wb.save(excelfile)
wb.close()
def main():
excel_info = []
site_info = {"192.168.158.2": {"username": "admin", "password": "admin"},
"192.168.158.3": {"username": "cisco", "password": "cisco"}}
#--- iteration for each site and get output
for host_ip in site_info:
username = site_info[host_ip]["username"]
password = site_info[host_ip]["password"]
hostname = ""
brand = ""
#-------- login
try:
with SSHClient() as client:
client.set_missing_host_key_policy(AutoAddPolicy())
client.connect(hostname=host_ip, username=username, password=password, look_for_keys=False, banner_timeout=60)
session = client.invoke_shell()
sleep(1)
#-- get data from session connect
data0 = session.recv(65535).decode('utf-8')
print(data0)
#-- create logfile and write
logfilename = f"log_{host_ip}.txt"
f = open(logfilename, "w")
f.write(data0.replace("\n",""))
f.close()
#-- extract data for get hostname
re_hostname = re.compile(r":(.*)#")
if re_hostname.search(data0):
hostname = re_hostname.search(data0).group(1)
#-- check brand by check last line returned by ssh
#-- last_line Nokia = A:NOKIA7750SR#
#-- last_line Cisco = RP/0/0/CPU0:Cisco_IOS_XRv#
last_line = data0.split("\n")[-1]
if "/CPU" in last_line:
brand = "Cisco"
else:
brand = "Nokia"
#-- send command line and get output
data1 = ""
data2 = ""
if hostname and brand == "Nokia":
waitstring = f"{hostname}#"
data1 = my_readstring("environment no more", session, waitstring, 30)
print(data1)
data2 = my_readstring("show port", session, waitstring, 30)
print(data2)
#-- extract data. similar to topic 5.2 Python_read_write_csv.py
re_nokia_port = re.compile(r"(\d+\/\d+\/\d+)\s+(Up|Down).*(Up|Down)\s+")
for aline in data2.split("\n"):
if re_nokia_port.search(aline):
port = re_nokia_port.search(aline).group(1)
port = "'" + port
admin_state = re_nokia_port.search(aline).group(2)
port_state = re_nokia_port.search(aline).group(3)
# -- store to excel_info to write excel file in further
excel_row = [hostname, host_ip, port, admin_state, port_state]
excel_info.append(excel_row)
elif hostname and brand == "Cisco":
waitstring = f"{hostname}#"
data1 = my_readstring("terminal length 0", session, waitstring, 30)
print(data1)
data2 = my_readstring("show ipv4 interface brief", session, waitstring, 30)
print(data2)
# -- extract data. similar to topic 5.2 Python_read_write_csv.py
re_cisco_port = re.compile(r"(GigabitEthernet[\d|\/]+)\s+\S+\s+(\w+)\s+(\w+)")
for aline in data2.split("\n"):
if re_cisco_port.search(aline):
port = re_cisco_port.search(aline).group(1)
admin_state = re_cisco_port.search(aline).group(2)
port_state = re_cisco_port.search(aline).group(3)
# -- store to excel_info to write excel file in further
excel_row = [hostname, host_ip, port, admin_state, port_state]
excel_info.append(excel_row)
# -- appending to logfile. remove double newline
data1 = data1.replace("\n","")
data2 = data2.replace("\n","")
f = open(logfilename, "a")
f.write(data1)
f.write(data2)
f.close()
except Exception as e:
print(f"Exception = {e} Occurred")
#--- write to excel
write_excel(excel_info)
print("Successful")
main()
สรุปและใส่ credit นิดหน่อย
