摘要

  • 这篇文章把关注点从“IP 属于谁”继续推进到“接口是什么状态、邻居链路是否可信”。
  • 方案核心是先用 SNMP 补齐接口,再用 LLDP 自动生成拓扑,并保留人工确认这一层兜底机制。
  • 如果你希望 NetBox 支撑自动化研判,而不是只存事实,这篇就是关键过渡。

核心目标:让 NetBox 从“资产台账”升级为“研判基线系统”


一、背景

在自动化运维中心建设过程中,最初认为问题在于:

  • 规则不够复杂
  • 日志不够全面
  • 自动化脚本不够智能

但实践后发现真正问题是:

缺乏可被信任的网络事实基线


1.1 典型问题

  • 接口 Down → 不知道是否异常
  • 邻居丢失 → 无法关联拓扑
  • 自动化操作(如 undo shutdown)不敢执行

1.2 本质问题

NetBox 仅停留在:

text
资产台账层

而没有进入:

text
研判基线层

二、设计目标

本方案设定三个核心目标:

  1. 接口自动发现
  2. 拓扑自动生成
  3. 拓扑必须人工确认

2.1 最终流程

发现流程

text
SNMP 发现接口
      ↓
NetBox 自动创建接口
      ↓
LLDP 发现邻居
      ↓
自动创建 Cable(tag=lldp-auto)
      ↓
人工确认(删除 tag)

三、环境准备

3.1 基础要求

  • Python ≥ 3.8
  • NetBox ≥ 4.x
  • 设备类型:华为(VRP)

3.2 网络要求

接口与邻居发现网络边界

网络边界

设备需开启:

  • SNMP
  • LLDP

3.3 Python 依赖

bash
pip install pysnmp pynetbox python-dotenv tqdm

3.4 环境变量

bash
NETBOX_URL=http://netbox-url
NETBOX_TOKEN=netbox-token
SNMP_COMMUNITY=public

四、步骤一:SNMP 自动发现接口

4.1 目标

自动补齐 NetBox 中所有设备接口信息


4.2 脚本:1_snmp_interfaces.py

python
#!/usr/bin/env python3
import os
import asyncio
from pysnmp.hlapi.asyncio.cmdgen import walkCmd, SnmpEngine, CommunityData, UdpTransportTarget, ContextData, ObjectType, ObjectIdentity
import pynetbox
from dotenv import load_dotenv
from tqdm import tqdm

load_dotenv()

NB = pynetbox.api(os.getenv("NETBOX_URL"), token=os.getenv("NETBOX_TOKEN"))
COMMUNITY = os.getenv("SNMP_COMMUNITY")

IF_NAME = '1.3.6.1.2.1.31.1.1.1.1'
IF_ALIAS = '1.3.6.1.2.1.31.1.1.1.18'
IF_ADMIN = '1.3.6.1.2.1.2.2.1.7'

async def snmp_walk(ip, oid):
    result = {}
    snmp_engine = SnmpEngine()
    iterator = walkCmd(
        snmp_engine,
        CommunityData(COMMUNITY),
        UdpTransportTarget((ip, 161), timeout=5, retries=2),
        ContextData(),
        ObjectType(ObjectIdentity(oid)),
        lexicographicMode=False
    )
    async for errorIndication, errorStatus, errorIndex, varBinds in iterator:
        for varBind in varBinds:
            idx = str(varBind[0]).split('.')[-1]
            result[idx] = str(varBind[1])
    return result

async def process_device(dev):
    if not dev.primary_ip:
        return
    ip = str(dev.primary_ip.address).split('/')[0]
    names = await snmp_walk(ip, IF_NAME)
    aliases = await snmp_walk(ip, IF_ALIAS)
    admins = await snmp_walk(ip, IF_ADMIN)

    for idx, if_name in names.items():
        if NB.dcim.interfaces.filter(device_id=dev.id, name=if_name):
            continue

        NB.dcim.interfaces.create(
            device=dev.id,
            name=if_name,
            type="1000base-t",
            enabled=(admins.get(idx) == '1'),
            description=aliases.get(idx, "")
        )

async def main():
    devices = NB.dcim.devices.filter(status="active")
    for dev in tqdm(devices):
        await process_device(dev)

if __name__ == "__main__":
    asyncio.run(main())

4.3 执行方式

bash
python3 1_snmp_interfaces.py

后台运行:

bash
nohup python3 1_snmp_interfaces.py > log.log 2>&1 &

4.4 执行效果

  • 自动补齐接口
  • 明确接口期望状态
  • 为拓扑构建打基础

五、步骤二:LLDP 自动生成拓扑

5.1 设计思路

LLDP ≠ 真实拓扑

因此采用:

text
自动发现 → 自动建模 → 人工确认

5.2 标记机制

text
tag = lldp-auto

含义:

  • 有 tag → 未确认
  • 无 tag → 已确认

5.3 脚本:2_snmp_lldp_auto_cable.py

python
#!/usr/bin/env python3
import os
from pysnmp.hlapi import *
import pynetbox
from dotenv import load_dotenv
from tqdm import tqdm

load_dotenv()

NB = pynetbox.api(os.getenv("NETBOX_URL"), token=os.getenv("NETBOX_TOKEN"))
COMMUNITY = os.getenv("SNMP_COMMUNITY")
AUTO_TAG = "lldp-auto"

def ensure_tag():
    tag = NB.extras.tags.get(name=AUTO_TAG)
    if not tag:
        tag = NB.extras.tags.create(name=AUTO_TAG, slug=AUTO_TAG)
    return tag

def snmp_walk(ip, oid):
    result = {}
    for (_, _, _, varBinds) in nextCmd(
        SnmpEngine(),
        CommunityData(COMMUNITY),
        UdpTransportTarget((ip, 161)),
        ContextData(),
        ObjectType(ObjectIdentity(oid)),
        lexicographicMode=False
    ):
        for varBind in varBinds:
            idx = str(varBind[0]).split('.')[-1]
            result[idx] = str(varBind[1])
    return result

tag = ensure_tag()

devices = NB.dcim.devices.filter(status="active")

for dev in tqdm(devices):
    if not dev.primary_ip:
        continue

    ip = str(dev.primary_ip.address).split('/')[0]

    local_ports = snmp_walk(ip, '1.0.8802.1.1.2.1.3.7.1.3')
    remote_names = snmp_walk(ip, '1.0.8802.1.1.2.1.4.1.1.9')
    remote_ports = snmp_walk(ip, '1.0.8802.1.1.2.1.4.1.1.7')

    for idx, local_if in local_ports.items():
        if idx not in remote_names:
            continue

        dev_b = NB.dcim.devices.get(name=remote_names[idx])
        if not dev_b:
            continue

        intf_a = NB.dcim.interfaces.get(device_id=dev.id, name=local_if)
        intf_b = NB.dcim.interfaces.get(device_id=dev_b.id, name=remote_ports.get(idx))

        if not intf_a or not intf_b:
            continue

        if intf_a.cable or intf_b.cable:
            continue

        NB.dcim.cables.create(
            tags=[tag.id],
            a_terminations=[{"object_type": "dcim.interface", "object_id": intf_a.id}],
            b_terminations=[{"object_type": "dcim.interface", "object_id": intf_b.id}]
        )

六、步骤三:人工确认拓扑

路径:

text
NetBox → DCIM → Cables → filter tag=lldp-auto

操作:

  • 正确 → 删除 tag
  • 错误 → 删除或修改

七、最终效果

7.1 研判能力提升

可以明确判断:

  • 接口是否应 Up
  • 是否存在对端
  • 是否单边故障

7.2 自动化安全边界

text
只对“已确认拓扑”执行自动化

7.3 NetBox 角色升级

从:

text
资产记录工具

升级为:

text
自动化决策输入源 + 安全控制中心

八、结语

自动化的难点不在脚本,而在:

text
事实是否可信,边界是否清晰

本方案本质解决:

将“网络真实结构”转化为“机器可信数据”

只有在这个基础上:

自动化中心才真正具备落地能力