首页 / 资讯中心 / 文章详情

Python正则+pydantic解析ASA防火墙回显,自动化生成设备台账

Python正则+pydantic解析ASA防火墙回显,自动化生成设备台账 ★ FEATURED ARTICLE
干网络运维的同学应该都有这种体验月底交台账领导让你把全网设备的软件版本、存活时间、接口状态统计一遍。你登录设备一条条敲命令把回显复制到文本文件里然后打开Excel手工整理。几十台设备干一晚上几百台设备直接怀疑人生。这个项目就是专门解决这个问题的——用Python的re库把采集到的ASA防火墙原始回显抽成结构化数据再用pydantic做模型校验最后输出成JSON、CSV台账几秒钟搞定。这套方案适合两类人一类是经常跟网络设备打交道、手里攒了一堆命令回显不知道怎么处理的运维工程师另一类是刚开始接触数据清洗的Python学习者。re库处理文本就像用手术刀做解剖pydantic则像一个严格的质检员两者配合起来能把脏乱差的原始文本变成干净、可靠、能直接入库的数据。下面我把整个思路和实操细节完整过一遍。1. 这个项目要解决什么问题1.1 采集到的ASA设备信息是什么样先看最原始的输入长什么样。ASA防火墙不像现在的很多设备支持结构化输出大部分时候你要的数据都藏在一堆人读起来很友好的文本里。从设备上敲show version回显大概是这样Adaptive Security Appliance Software Version 9.12(4)31 Device Manager Version 7.12(1)152 Compiled on Tue Apr 9 19:54:08 PDT 2019 by builders System image file is disk0:/asa912-4-smp-boot.img Config file at boot was startup-config asaserver up 120 days 5 hours Hardware: ASA5516, 8192 MB RAM, CPU Pentium II 1500 MHz, 1 CPU (2 cores) Model Id: ASA5516再敲show running-config接口配置是这种风格interface GigabitEthernet0/0 nameif outside security-level 0 ip address 192.0.2.10 255.255.255.0 ! interface GigabitEthernet0/1 nameif inside security-level 100 ip address 10.0.0.1 255.255.255.0 ! interface GigabitEthernet0/2 shutdown ! interface Management0/0 nameif management security-level 100 ip address 192.0.2.254 255.255.255.0 !如果你采集的是show interface格式又完全不一样Interface GigabitEthernet0/0 outside, is up, line protocol is up Hardware is GigabitEthernet, address is 0011.2233.4455 Internet address is 192.0.2.10/24 MTU 1500 bytes, BW 1000000 Kbit这些文本本身没有错误但它们不是结构化数据。你要提取软件版本号“接口名称”“IP地址”这些字段就得自己写解析逻辑。这里有两个核心难点一是各类命令的回显格式差异很大正则要针对不同的行结构分别处理二是同一条命令在不同版本、不同型号的设备上可能略有差别解析规则必须足够鲁棒。1.2 为什么选re库和pydantic这对组合处理文本的方案其实不少。有人用字符串split硬切有人用现成的网络设备解析库还有人用TextFSM模板引擎。我这次选re加pydantic是因为这套组合在灵活度和数据可靠性之间取得了最好的平衡。re库是Python自带的正则模块对付ASA回显这种结构相对固定、但格式又有点小变化的文本非常合适。正则表达式可以精确描述我要找的是什么样的行比如show interface里的状态行特征就是接口名 引号里的别名 up/down状态用一行正则就能抓全。pydantic解决的是另一个问题正则提取出来的是字符串类型全都是str但你要的接口安全级别是整数、IP地址需要校验合法性、接口列表里还可能有缺失字段。这些数据质量问题如果靠手写if判断会写出一堆又臭又长的代码pydantic用模型的字段定义和校验器就把它规范掉了。这对组合还有一个很实际的好处re负责从无结构文本提取半结构化字段pydantic负责从半结构化字典变成强类型模型两者职责完全分离出问题的时候定位特别快。正则没匹配上是re层的问题字典有了但值不合法是pydantic层的问题谁也不甩锅。2. 整体方案设计从原始文本到结构化数据2.1 数据流向与分层设计整个解析流程我分成四层每一层只干一件事方便后续维护和扩展原始文本文件 - 文本预处理 - 正则提取字典 - pydantic校验实例 - 导出JSON/CSV第一层是文本预处理。这一步特别容易被忽略但特别重要。从不同环境采集回来的文件可能有\r\n和\n混用的问题可能有BOM头还可能有采集工具自动加的行号。先把这些脏东西统一处理掉后面的正则才能稳定工作。我会先做一次全局清洗def clean_raw_text(text: str) - str: # 去掉Windows换行符残留 text text.replace(\r\n, \n) # 去BOM if text.startswith(\ufeff): text text[1:] # 去掉行首行尾的空白字符 lines [line.strip() for line in text.split(\n)] return \n.join(lines)第二层是正则提取。我的做法是先切块、再逐块细提而不是用一串巨型正则匹配整个文件。比如运行配置里每个接口配置块以!结尾那就先把整段配置按!切成多个块再对每个块判断它是不是接口块、有没有nameif、有没有IP地址。这种分层处理的好处是逻辑清晰哪个块解析失败也不影响其他块。第三层是pydantic校验。正则提取出来的字典字段值都是字符串pydantic模型定义好字段类型后会自动完成类型转换和合法性检查。比如security-level定义了int类型字符串100会被自动转成整数100如果设备上写了security-level abc这种非法值pydantic直接抛ValidationError我们就能在批量处理时精准捕获这条脏数据。第四层是导出。单台设备解析完成后是一个ASAConfig模型实例调用model_dump()转成字典再写入JSON或CSV。批量处理几十个文件时每台设备的结果单独存一份JSON同时汇总所有设备的关键字段形成一个CSV台账。2.2 配置块切分先用正则做好粗加工配置切分是整个项目里最需要细心的一步。ASA运行配置里用单独一行的!作为段落分隔符这个符号本身没有含义就是排版用的分节符。用Python的re.split可以很方便地切分import re segments re.split(r^!\s*$, running_config, flagsre.MULTILINE)这里有个容易踩的坑。如果不加re.MULTILINE^和$匹配的是整个字符串的开头和结尾而不是每一行的开头和结尾那样根本切不动。加了MULTILINE之后^!才能匹配到行首的!。另外\s*$是考虑到有些!后面可能有空格或\r残留。切出来的每个segment就是一段独立配置。接下来对每段做分类我针对常见场景写了三个判断条件优先级从高到低以interface开头接口配置块继续细解析包含hostname设备主机名配置包含access-listACL配置这个项目暂时不深入。接口配置块细解析时用findall而不是search因为一个块里可能有多条关键信息比如nameif、security-level、ip address、vlan。把这些字段的正则收集到一起逐条匹配最后合成字典。2.3 pydantic模型给数据上规矩模型设计的核心思想是该有的字段必须有不该有的格式错误要拦住。我定义的接口信息模型长这样from typing import List, Optional from pydantic import BaseModel, Field, field_validator class InterfaceInfo(BaseModel): name: str nameif: Optional[str] None security_level: Optional[int] None ip_address: Optional[str] None netmask: Optional[str] None vlan: Optional[int] None status: Optional[str] None field_validator(ip_address) classmethod def check_ip(cls, v): if v is None: return v parts v.split(.) if len(parts) ! 4: raise ValueError(f非法IP地址: {v}) for part in parts: if not part.isdigit() or not 0 int(part) 255: raise ValueError(f非法IP地址: {v}) return v把字段定义为Optional是故意的。一台物理接口如果处于shutdown状态通常没有IP地址接口块里自然就不存在ip address这一行。如果模型把ip_address设为必填那所有down掉的接口都会校验失败批量处理就寸步难行。用Optional[str] None缺字段时值就是None语义上完全表达清楚。顶层模型把接口列表和设备基本信息装在一起class ASAConfig(BaseModel): hostname: str software_version: str uptime_days: int hardware_model: Optional[str] None serial_number: Optional[str] None interfaces: List[InterfaceInfo] Field(default_factorylist)这里的Field(default_factorylist)必须用default_factory而不是default[]。这是一个经典的Python可变默认值陷阱直接用空列表做默认值的话所有模型实例会共享同一个列表对象一个实例往里加接口其他实例的接口列表也会跟着变。pydantic在新版本里其实会在内部处理这个问题但养成用default_factory的习惯能避免很多莫名其妙的bug。3. re库实战解析ASA信息提取的几个核心正则3.1 版本与硬件信息的抓取show version的文本结构相对规整一行一个字段用简单的行匹配就能解决。软件版本那行最有辨识度的关键词是Software Version后面跟的是版本号。版本号里既有数字、括号还有小数点正则要写成这样version_match re.search(rSoftware Version\s([\d.()]), version_text)这里字符集[\d.()]限定了版本号只由数字、点和括号组成。如果设备软件版本是9.12(4)31匹配到的就是完整字符串。不用\d\.\d这种硬编码结构是因为不同版本号格式差异很大有的带括号补丁号有的不带宽松匹配反而更稳。存活时间那行的特征是up 天数 小时数uptime_match re.search(rup\s(\d)\sdays?\s(\d)\shours?, version_text) days int(uptime_match.group(1)) hours int(uptime_match.group(2))这里用了days?和hours?这种问号后缀写法意思是前面的字符s出现0次或1次。因为英文里1 day不带s而120 days带s如果不加这个问号匹配1 day就会失败。这种细节就是正则的鲁棒性来源处理真实设备回显时特别管用。硬件型号那行是Hardware: ASA5516这种格式字段之间空格数量不固定用\s做分隔符hardware_match re.search(rHardware:\s(\S), version_text) asamodel hardware_match.group(1) if hardware_match else None\S匹配连续非空白字符正好合适。这里我没加re.IGNORECASE因为ASA默认输出固定大写加了反而可能误匹配到注释里的小写单词。3.2 接口配置段的拆分与字段提取接口配置块的拆分在前面提过先用re.split按!切块然后对每个接口块提取字段。每个字段对应一个正则我把它们整理成了一张表字段正则表达式说明接口名^interface\s(\S)行首的interface关键字后跟接口名nameif^\s*nameif\s(\S)注意行首可能有空格缩进security-level^\s*security-level\s(\d)安全级别是0到100的整数IP地址ip address\s(\d\.\d\.\d\.\d)\s(\d\.\d\.\d\.\d)地址和掩码是两段vlan^\s*vlan\s(\d)子接口上才有的字段提取逻辑用search逐条查找找到就取group找不到就跳过。代码写出来就是一组if-elif结构每段代码都很短一眼能看懂def parse_interface_block(block: str) - dict: info {} iface_match re.search(r^interface\s(\S), block, re.MULTILINE) if not iface_match: return {} info[name] iface_match.group(1) nameif_match re.search(r^\s*nameif\s(\S), block, re.MULTILINE) if nameif_match: info[nameif] nameif_match.group(1) sec_match re.search(r^\s*security-level\s(\d), block, re.MULTILINE) if sec_match: info[security_level] int(sec_match.group(1)) ip_match re.search( rip address\s(\d\.\d\.\d\.\d)\s(\d\.\d\.\d\.\d), block ) if ip_match: info[ip_address] ip_match.group(1) info[netmask] ip_match.group(2) vlan_match re.search(r^\s*vlan\s(\d), block, re.MULTILINE) if vlan_match: info[vlan] int(vlan_match.group(1)) return info如果采集的是show interface的回显而不是运行配置状态字段的提取方式不太一样。状态信息都在接口描述行里格式是Interface GigabitEthernet0/0 outside, is up, line protocol is up。提取时我写成status_match re.search( rInterface\s(\S)\s([^]*),\sis\s(up|down),\sline protocol is\s(up|down), iface_fa_text )这里([^]*)匹配引号里的接口别名[^]*的意思是除双引号外的任意字符能防止贪婪匹配跨到下一行去。状态枚举(up|down)直接限定了可选值比\w更严谨像administratively down这种长状态如果出现这个正则也能留出口子后续按需扩展。3.3 兼容性细节多行匹配、空白符、大小写正则写多了就会发现真正让人头疼的不是匹配规则本身而是各种边界情况。我总结了自己踩过坑的几个点。第一是re.MULTILINE别忘带。很多行首匹配的正则不指定MULTILINE的话^匹配的是整段文本的开头不是每一行的开头。这个问题在解析接口配置时尤其致命因为接口块内部有很多缩进行的字段^如果只匹配字符串开头除了块的第一行以外什么都匹配不到。第二是空白符处理。设备回显里经常出现关键字和值之间多个空格的情况比如nameif outside可能比nameif outside多一个空格。所以字段之间用\s而不是单个空格匹配能大幅提高鲁棒性。但要注意\s也会匹配换行符如果在一行正则里用^\s*nameif前面的\s*可能把上一行的换行吞掉导致匹配到错误的位置。我一般会在行首锚定场景下用[ \t]*来明确只匹配空格和制表符不匹配换行。第三是大小写问题。ASA设备的输出默认大写但采集工具或者人工转储过程中可能改变大小写。如果要把解析器做成通用工具统一在正则上不加IGNORECASE、在匹配前把文本upper()一下是更可控的做法。直接在每条正则上加re.IGNORECASE也行但那样会让调试时更难判断到底匹配了什么。4. pydantic模型设计与校验逻辑4.1 字段类型自动转换的妙用pydantic最直观的价值是类型转换。正则提取出来的security_level是字符串100模型里定义成intpydantic会自动帮我们转成整数100。这在导出CSV时省了很多事——CSV不需要带引号的数字字符串直接用整数写进去更干净。但自动转换不是万能的。如果设备配置里出现security-level abc这种数据字符串abc转不成整数pydantic会抛出ValidationError。这其实是好事它帮你主动暴露了数据质量问题。在真实项目中这种异常字段往往意味着设备配置有问题要么是采集不完整要么是有人误配置总归值得你去核查。4.2 自定义validator的几种实用场景除了类型转换pydantic还支持自定义校验器。我实际用到的主要有三种场景。场景一是IP地址合法性校验。前面接口模型里的check_ip就是一个例子。其实Python标准库里有更简洁的写法import ipaddress field_validator(ip_address) classmethod def check_ip(cls, v): if v is None: return v ipaddress.ip_address(v) # 非法IP会抛异常 return vipaddress模块是标准库自带不需要额外安装。它不仅能校验IPv4还能区分IPv6比手工拆点分更可靠。场景二是跨字段校验。比如一个接口如果配置了ip address理论上它不应该同时出现在shutdown接口上。这种字段间关联关系用model_validator处理from pydantic import model_validator model_validator(modeafter) def check_ip_not_on_shutdown(self): if self.status down and self.ip_address: raise ValueError(f接口{self.name}处于down状态但配置了IP地址) return self场景三是值域归一化。有些字段不同设备写法不一样比如接口别名有的设备写成outside有的写成OUTSIDE。可以在校验器里统一转成小写后续统计时就不怕大小写不一致导致分组出错。4.3 嵌套模型与可选字段的处理接口列表用的是嵌套模型ASAConfig.interfaces是List[InterfaceInfo]。嵌套模型的好处是校验逻辑也能嵌套每个接口独立校验接口的字段错误只会影响这一个接口不会导致整台设备解析失败。实际批量解析时我不建议让一个错误直接中断整个流程。更好的模式是收集错误继续处理。比如某台设备有10个接口其中1个接口缺失IP地址字段导致校验失败我希望其他9个接口正常入库只把出问题的那个记录下来。这个需求用pydantic的ValidationError配合循环可以实现。先把所有接口字典逐个转成InterfaceInfo实例失败的放进错误列表成功的放进最终结果interfaces [] errors [] for iface_dict in iface_dict_list: try: interfaces.append(InterfaceInfo.model_validate(iface_dict)) except ValidationError as exc: errors.append({interface: iface_dict.get(name), detail: exc.errors()})这里用了pydantic v2的model_validate方法。如果你用的还是v1版本对应的方法是parse_obj两者在参数传入方式上有差异升级到v2之后旧代码会报警告。项目新写的话直接用v2语法就好。5. 完整实操从文本到JSON的一键流水线5.1 解析函数组装把前面所有部件组装起来核心就是一个parse_asa_text函数。它接收原始文本返回一个ASAConfig实例。整体流程是先全局清洗文本再从show version回显里提取版本、硬件、存活时间再切分配置块提取接口列表最后用ASAConfig.model_validate组装模型。def parse_asa_text(full_text: str) - ASAConfig: text clean_raw_text(full_text) # 提取主机名 host_match re.search(r^hostname\s(\S), text, re.MULTILINE) hostname host_match.group(1) if host_match else 未知设备 # 提取版本信息 ver_match re.search(rSoftware Version\s([\d.()]), text) software_version ver_match.group(1) if ver_match else 未知版本 # 提取存活时间 uptime_match re.search(rup\s(\d)\sdays?\s(\d)\shours?, text) uptime_days int(uptime_match.group(1)) if uptime_match else 0 # 提取硬件型号 hw_match re.search(rHardware:\s(\S), text) hardware_model hw_match.group(1) if hw_match else None # 切分配置块 segments re.split(r^!\s*$, text, flagsre.MULTILINE) iface_dict_list [] for seg in segments: if seg.strip().startswith(interface): iface_dict_list.append(parse_interface_block(seg)) return ASAConfig( hostnamehostname, software_versionsoftware_version, uptime_daysuptime_days, hardware_modelhardware_model, interfacesiface_dict_list, )这里有一个设计取舍切分配置块的时候我按整段文本一次性切分然后过滤出接口块。如果采集回来的文本里同时有show version和show running-config的内容版本信息那部分会被切进某个段里。好在段分类逻辑足够严格只有interface开头的段才进入接口解析所以版本段会被自然忽略。5.2 批量处理多台设备文件项目真正要处理的是一批设备不是一个文件。我用pathlib遍历指定目录下的所有文本文件逐个解析结果统一保存from pathlib import Path import json input_dir Path(./asa_raw_texts) output_dir Path(./asa_parsed) output_dir.mkdir(exist_okTrue) results [] for file_path in input_dir.glob(*.txt): text file_path.read_text(encodingutf-8, errorsignore) try: device parse_asa_text(text) results.append(device) # 每台设备单独存一份JSON output_file output_dir / f{device.hostname}.json output_file.write_text( json.dumps(device.model_dump(), ensure_asciiFalse, indent2), encodingutf-8 ) except ValidationError as exc: print(f解析失败: {file_path.name} - {exc.errors()})读取文件时用了errorsignore这是处理编码不确定的文本文件常用的手法。有的设备采集工具导出的文件是GBK编码直接用UTF-8读会抛UnicodeDecodeError错误。errorsignore会把无法解码的字节丢弃虽然会损失极少数字符但总比整个文件读不进来强。如果你的环境能确定编码尽量指定正确的编码不要无脑忽略错误。批量处理还有一个关键点异常隔离。ValidationError只捕获了pydantic的校验错误但正则匹配失败不会抛异常而是返回None或空字典。我在parse_asa_text里给版本号等字段都设置了默认值未知版本就是为了防止某个字段没匹配上时直接把整个流程搞挂。宁可字段值为未知也不能让一条脏数据拖垮整批处理。5.3 结果输出与异常隔离解析完成的ASAConfig实例转成字典之后用json.dumps写入文件。如果想要一份汇总的CSV台账供表格查看可以用Python标准库的csv模块import csv with open(output_dir / summary.csv, w, newline, encodingutf-8-sig) as f: writer csv.DictWriter(f, fieldnames[hostname, software_version, uptime_days, interface_count]) writer.writeheader() for device in results: writer.writerow({ hostname: device.hostname, software_version: device.software_version, uptime_days: device.uptime_days, interface_count: len(device.interfaces), })CSV文件用utf-8-sig编码写入是为了兼容Excel。utf-8-sig会在文件头部加BOM标记Excel打开时才不会出现中文乱码。这个细节是给那些拿到数据还要交给不懂技术的同事看的场景准备的省掉一遍无谓的来回沟通。如果想把接口明细也打进CSV可以针对接口列表单独再写一个文件。每个接口一行字段包括设备名、接口名、别名、IP、掩码、安全级别这样一个文件就能直接用于端口台账的核对。6. 常见问题与排查技巧实录6.1 正则贪婪匹配导致跨块吞数据这是我最先踩的坑。写接口名称匹配时一开始我用了rinterface\s(.)想着用.匹配完整接口名。结果.是贪婪的它会一直匹配到行尾如果接口配置块里interface后面还有别的说明文字就会把不该吞的内容也吞进去。后来我改成rinterface\s(\S)问题解决。\S只匹配非空白字符遇到空格就停下来。这种宁可匹配少一点不要匹配过头的思路贯穿了整个项目。凡是能精确限定字符集的就不用宽泛的.能用字符类[\d.()]的就不用\w。排查贪婪匹配问题有个快速方法在正则的末尾加一个已知的锚定字符比如行尾$或者后面的固定格式。如果加了锚定之后匹配结果变了说明之前确实匹配过头了。6.2 接口状态字段格式不一致show interface的状态行在不同设备上有细微差异。有的是is up, line protocol is up有的是is administratively down, line protocol is down还有的老设备会写成is up, line protocol is up (connected)。我的应对策略是尽量使用更通用的正则比如把状态匹配拆成两个小部分admin_state re.search(ris\s([a-z ]?),\s*line protocol, iface_line, re.IGNORECASE) proto_state re.search(rline protocol is\s([a-z ]?)\s*$, iface_line, re.IGNORECASE)这种写法用([a-z ]?)加问号做了非贪婪匹配意思是尽量少匹配直到遇到后面的逗号和line protocol。这样即使状态描述里有额外单词也能把核心状态提取出来。提取出来的状态值如果包含down就认为接口是down状态再统一归一化存储。我建议不要试图用一条正则覆盖所有格式变体而是准备两三条候选正则按顺序尝试。第一条用最严格的格式匹配不上再用宽松的兜底正则这样既能保证精确匹配优先又能应对异常格式。6.3 pydantic校验失败的定位方法pydantic v2的ValidationError异常对象有errors()方法返回的是一个列表里面每一项都包含loc字段和msg字段。loc告诉你是哪个字段出了问题msg告诉你是哪条校验规则没通过。我在批量解析时遇到过一条报错接口GigabitEthernet0/2的ip_address是192.168.1.最后一段是空的。check_ip校验器发现part为空字符串isdigit()返回False果断抛错。当时如果没有pydantic这层校验这条脏数据就会悄悄写进JSON等真正用到的时候才发现是坏的。定位校验错误时推荐在代码里打印结构化的errors()而不是只打印异常的str()。str()会把所有错误信息揉成一大段难以阅读的文本而errors()返回的字典列表结构清晰字段名和错误原因一目了然排查效率高得多。6.4 其他几个容易忽略的细节第一是文件编码问题。设备导出的文本除了UTF-8和GBK还可能带UTF-8 BOM。clean_raw_text里我专门处理了\ufeff就是这个原因。BOM如果不处理它会附着在第一个字段的字符串前面导致hostname值变成\ufeffASAServerpydantic校验时完全看不出来但在最终数据里是个隐藏炸弹。第二是空文件或采集失败的文件。批量处理时可能有个别文件内容为空或者只有一句错误提示。parse_asa_text里的默认值兜底机制保证了解析不会崩溃但生成的结果语义上可能不太对。我会在批量脚本里增加一个简单判断如果解析出的software_version是未知版本就把这个文件标记为需要人工复核。第三是设备重启后的存活时间重置。uptime_days如果直接存整数重启后会归零这在统计上是正确的但如果你要做设备最长稳定运行时间的排名需要保留原始采集时间点。我一般在导出的JSON里额外加一个collected_at字段记录本次采集的时间戳这样每次采集的记录都能单独追溯。7. 一点使用体会这套re加pydantic的组合我后来又用在了解析其他设备的回显上流程基本没大改只是换了正则和模型字段。最深的体会是正则负责提取pydantic负责约束两者缺一不可。只用正则的话提取出来的数据是散装字符串质量参差不齐后面用的时候还得反复做防御性判断只用pydantic的话没有正则帮我们从回显里抽取字段模型再严谨也没数据可校验。还有一个经验是先切块、再细提永远比一条大正则通吃好维护。大正则写的时候很爽但三个月后自己回来看都费劲更别说交接给同事。切成小块之后每个正则只负责一件事改起来也只动局部回归测试的代价小得多。如果你手里正好有一堆设备回显要整理我建议不要急着手工复制粘贴先用这套思路跑一个小脚本试试。从单台设备的两个命令回显开始把解析函数写顺了再扩展到批量处理。等脚本稳定跑通了你会发现原来需要一下午的台账工作现在真的只需要几秒钟。
阅读完成 · 觉得有帮助?
咨询建站