自動化運維實驗

一、實驗拓撲

二、實驗目的

三、實驗步驟

實驗思路：

代碼部分：

四、實驗結果：

一、實驗拓撲

二、實驗目的

利用python腳本，在本地，或者虛擬機里實現，設備CRC數量統計，并輸出成表格

三、實驗步驟

實驗開始之前先搭好環境，測試無誤再開始

實驗思路：

利用ip.txt，存放需要檢查的設備IP

再一臺一臺登錄，匹配并記錄接口名，和CRC數量，最終再輸出保存

代碼部分：

#!/usr/bin/env python3# -*- coding: utf-8 -*-import argparseimport osimport reimport timeimport socketimport getpassimport paramikoimport pandas as pdfrom paramiko import SSHException, AuthenticationExceptionDEFAULT_IP_FILE = 'ip.txt'DEFAULT_OUTPUT = 'crc_results.xlsx'DEFAULT_COMMAND = 'dis interface'DEFAULT_CONNECT_TIMEOUT = 5DEFAULT_CMD_TIMEOUT = 30MORE_PROMPTS = [b'--More--', b'---- More ----', b'--More--', b'More:', b'== More ==', b'--More--', b'\x1b\[K--More--']def read_ip_list(path):ips = []if not os.path.exists(path):print(f"[!] IP file not found: {path}")return ipswith open(path, 'r', encoding='utf-8') as f:for raw in f:line = raw.strip()if not line:continueif line.startswith('#'):continueparts = re.split(r'[\s,]+', line)if parts and parts[0]:ips.append(parts[0])return ipsdef recv_with_paging(shell, timeout=DEFAULT_CMD_TIMEOUT):"""從交互 shell 中讀取輸出，自動處理分頁（遇到 --More-- 等發送空格繼續）。返回解碼后的文本。"""buf = b''start = time.time()last_recv = time.time()shell.settimeout(1.0)while True:try:if shell.recv_ready():chunk = shell.recv(65536)if not chunk:breakbuf += chunklast_recv = time.time()# 檢查是否包含常見分頁提示（原始字節）for mp in MORE_PROMPTS:if mp in chunk:try:shell.send(' ')except Exception:pass# 也處理常見 ASCII 控制字符后跟 More 文本（部分設備會帶 escape）# 繼續循環讀取else:time.sleep(0.1)except socket.timeout:# no data right nowpassexcept Exception:# 其他異常則跳出breakif (time.time() - start) > timeout:break# 當已有數據并且很久沒讀到新數據，認為輸出結束if buf and (time.time() - last_recv) > 1.2:breaktry:return buf.decode(errors='ignore')except:return buf.decode('utf-8', 'ignore')def run_command_interactive(client, command, cmd_timeout=DEFAULT_CMD_TIMEOUT):shell = client.invoke_shell()time.sleep(0.2)# 嘗試關閉分頁# 兼容多廠商：terminal length 0, screen-length disable, page-off 等try:shell.send('terminal length 0\n')time.sleep(0.15)shell.send('screen-length 0 temporary\n')time.sleep(0.15)shell.send('page off\n')time.sleep(0.15)except Exception:pass# 清空 bannertry:if shell.recv_ready():_ = shell.recv(65536)except Exception:pass# 發送命令并讀取，自動處理分頁try:shell.send(command + '\n')except Exception:passoutput = recv_with_paging(shell, timeout=cmd_timeout)try:shell.close()except Exception:passreturn outputdef split_interface_blocks(output):"""更嚴格的 H3C 專用分割：- 識別常見 H3C 前綴：GigabitEthernet, Ethernet, Ten-GigabitEthernet, Eth-Trunk,Vlan-interface / Vlanif, Loopback, FortyGigabitEthernet 等。- 要求接口名包含數字，避免把 'Flow-control' 等字段當接口頭。返回列表 (iface_name, block_text)"""lines = output.splitlines()blocks = []cur_iface = Nonecur_lines = []# 列出常見的接口前綴（可按需擴展）iface_prefixes = [r'GigabitEthernet', r'GigEth', r'Ten-GigabitEthernet', r'FortyGigabitEthernet',r'Ethernet', r'Eth-Trunk', r'Vlan-interface', r'Vlanif', r'Loopback', r'VEth', r'Veth']prefix_pat = r'(' + '|'.join(iface_prefixes) + r')'# 可能的 header 樣式：# "GigabitEthernet1/0/1 current state : UP"# "Interface: GigabitEthernet1/0/1"# "GigabitEthernet1/0/1 is down"p1 = re.compile(r'^\s*(?:Interface[:\s]+)?' + prefix_pat + r'([A-Za-z0-9\/\-\._]*\d+)\b', re.IGNORECASE)# 另一種：全名連續（例如 "GigabitEthernet1/0/1 current state : UP"）p2 = re.compile(r'^\s*' + prefix_pat + r'([A-Za-z0-9\/\-\._]*\d+)\b.*(?:current state|is up|is down|UP|DOWN|administratively down)', re.IGNORECASE)for ln in lines:ln_str = ln.rstrip('\r\n')m = Nonefor p in (p1, p2):mm = p.match(ln_str)if mm:# 名字由 mm.group(1)+group(2) 或 mm.group(1)+group(2) 組成，取匹配到的完整接口名# 某些 pattern 捕獲了兩部分（前綴和后綴），拼接起來groups = [g for g in mm.groups() if g]name = ''.join(groups).strip()# 最終再確認包含數字if not re.search(r'\d', name):continuem = namebreakif m:name = m.rstrip(',:')if cur_iface is not None:blocks.append((cur_iface, '\n'.join(cur_lines)))cur_iface = namecur_lines = [ln_str]else:if cur_iface is not None:cur_lines.append(ln_str)else:cur_lines.append(ln_str)if cur_iface is not None:blocks.append((cur_iface, '\n'.join(cur_lines)))else:blocks = [('Unknown', '\n'.join(cur_lines).strip())] if cur_lines else [('Unknown', output.strip())]return blocksdef extract_crc_from_block(block_text):"""多種方式提取 CRC 數字：- '4 input errors, 2 CRC, 0 frame' -> 2- 'CRC: 2' or 'CRC 2' or '2 CRC' -> 2- 'input errors: 2, CRC: 2' etc.返回整數（若能解析）或 None。"""# 常見組合patterns = [re.compile(r'(\d+)\s+input errors.*?(\d+)\s+CRC', re.IGNORECASE | re.DOTALL),re.compile(r'input errors[:\s]*\d+[,;\s]*CRC[:\s]*(\d+)', re.IGNORECASE),re.compile(r'CRC[:\s]+(\d+)', re.IGNORECASE),re.compile(r'\b(\d+)\s+CRC\b', re.IGNORECASE),re.compile(r'CRC errors[:\s]*(\d+)', re.IGNORECASE),re.compile(r'CRC\s+(\d+)', re.IGNORECASE)]for rg in patterns:m = rg.search(block_text)if m:# 找最后一組數字（某些 pattern 有兩個捕獲組）for g in reversed(m.groups()):if g and g.isdigit():try:return int(g)except:pass# 如果找不到數字，但出現單詞 CRC，則標記為 Unknown (存在 CRC 詞但無數字)if re.search(r'\bCRC\b', block_text, re.IGNORECASE):return None? # Unknown count but CRC mentioned# 否則認為沒有 CRC（記錄為 0）return 0def check_one_device(ip, username, password, command, connect_timeout, cmd_timeout):client = paramiko.SSHClient()client.set_missing_host_key_policy(paramiko.AutoAddPolicy())try:client.connect(hostname=ip,username=username,password=password,timeout=connect_timeout,banner_timeout=connect_timeout,auth_timeout=connect_timeout)except AuthenticationException as e:return False, f'Authentication failed: {e}'except socket.timeout:return False, 'Connection timed out'except SSHException as e:return False, f'SSHException: {e}'except Exception as e:return False, f'Connection failed: {e}'try:output = run_command_interactive(client, command, cmd_timeout=cmd_timeout)except Exception as e:try:client.close()except:passreturn False, f'Failed to run command: {e}'try:client.close()except:passif not output:return True, []? # 沒有輸出也返回成功但沒有接口blocks = split_interface_blocks(output)results = []for iface, block in blocks:crc_val = extract_crc_from_block(block)# 規范輸出值：數字 -> string of number; None -> 'Unknown'; 0 -> '0'if crc_val is None:crc_str = 'Unknown'else:crc_str = str(int(crc_val))results.append({'interface': iface, 'crc': crc_str})return True, resultsdef main():parser = argparse.ArgumentParser(description='Check interfaces CRC on devices listed in ip.txt')parser.add_argument('-i', '--ipfile', default=os.environ.get('IP_FILE', DEFAULT_IP_FILE), help='IP list file (default ip.txt)')parser.add_argument('-u', '--username', help='SSH username (or set SSH_USER env)')parser.add_argument('-p', '--password', help='SSH password (or set SSH_PASS env) - warning: visible in process list')parser.add_argument('-c', '--command', default=os.environ.get('CRC_CMD', DEFAULT_COMMAND), help=f'Command to run on device (default: "{DEFAULT_COMMAND}")')parser.add_argument('-o', '--output', default=DEFAULT_OUTPUT, help='Output Excel filename')parser.add_argument('--connect-timeout', type=int, default=int(os.environ.get('SSH_CONNECT_TIMEOUT', DEFAULT_CONNECT_TIMEOUT)), help='SSH connect timeout seconds (default 5)')parser.add_argument('--cmd-timeout', type=int, default=int(os.environ.get('SSH_CMD_TIMEOUT', DEFAULT_CMD_TIMEOUT)), help='Command read timeout seconds (default 30)')args = parser.parse_args()ips = read_ip_list(args.ipfile)if not ips:print(f'[!] No IPs found in {args.ipfile}')returnusername = args.username or os.environ.get('SSH_USER') or input('SSH username: ')password = args.password or os.environ.get('SSH_PASS') or getpass.getpass('SSH password: ')all_rows = []failures = []print(f'[+] Loaded {len(ips)} IPs, connecting with timeout {args.connect_timeout}s, command timeout {args.cmd_timeout}s')for ip in ips:print(f'[{ip}] connecting...')success, data = check_one_device(ip, username, password, args.command, args.connect_timeout, args.cmd_timeout)if not success:print(f'[{ip}] FAILED: {data}')failures.append({'ip': ip, 'reason': data})continueparsed = dataif not parsed:print(f'[{ip}] OK, but no interface blocks parsed.')# still add a row to indicate device had no data? We skip adding rows for this device (user requested only IP/interface/crc)continueprint(f'[{ip}] OK, {len(parsed)} interfaces parsed.')for ent in parsed:all_rows.append({'ip': ip, 'interface': ent['interface'], 'crc': ent['crc']})# 寫 Excel：只包含 ip, interface, crc；另一個 sheet 記錄失敗df = pd.DataFrame(all_rows, columns=['ip', 'interface', 'crc'])df_fail = pd.DataFrame(failures, columns=['ip', 'reason'])with pd.ExcelWriter(args.output, engine='openpyxl') as writer:df.to_excel(writer, sheet_name='crc_results', index=False)df_fail.to_excel(writer, sheet_name='ssh_failures', index=False)print(f'[+] Done. Results saved to {args.output}')print(f'[+] Devices checked: {len(ips)}, interfaces rows: {len(all_rows)}, ssh failures: {len(failures)}')if __name__ == '__main__':main()