第305篇：NAPALM 统一网络设备管理接口

关键词

NAPALM、多厂商统一 API、配置管理、Getter 方法、配置替换、配置提交、网络自动化、Python 网络库

一、NAPALM 简介

1.1 什么是 NAPALM

NAPALM（Network Automation and Programmability Abstraction Layer with Multivendor support）是一个 Python 库，为网络设备提供 统一的跨厂商管理接口：

NAPALM 的设计理念：

Netmiko vs NAPALM 定位差异： ┌─ Netmiko：CLI 驱动层，发送命令 + 回显解析 ├─ NAPALM：抽象 API 层，统一数据模型 + 配置管理 └─ 两者可以互补使用

安装：

python -m pip install napalm

1.2 NAPALM 三大核心能力

NAPALM 提供三类操作：

  1. 数据收集（Getter）
  ┌─ get_facts()：设备基本信息
  ├─ get_interfaces()：接口状态
  ├─ get_interfaces_ip()：接口 IP
  ├─ get_lldp_neighbors()：LLDP 邻居
  ├─ get_bgp_neighbors()：BGP 邻居
  ├─ get_environment()：电源/风扇/温度
  ├─ get_arp_table()：ARP 表
  ├─ get_mac_address_table()：MAC 表
  ├─ get_route_to()：路由表
  ├─ get_ntp_peers()：NTP 配置
  └─ get_ntp_servers()：NTP 服务器

  2. 配置管理
  ┌─ load_replace_candidate()：加载替换配置
  ├─ load_merge_candidate()：加载合并配置
  ├─ compare_config()：比较配置差异
  ├─ commit_config()：提交变更
  ├─ discard_config()：丢弃候选配置
  └─ rollback()：回滚

  3. 验证与文件
  ├─ ping()：设备 ping
  ├─ traceroute()：路由追踪
  └─ config_control()：保存配置

二、数据收集实战

2.1 连接与获取设备信息

#!/usr/bin/env python3
# napalm_getter.py — 使用 Getter 收集设备数据

from napalm import get_network_driver
import json

# 1. 选择驱动
driver = get_network_driver("huawei_vrp")  # 华为 VRP

# 2. 创建设备对象
device = driver(
    hostname="192.168.1.1",
    username="admin",
    password="admin123",
    optional_args={"port": 22},
)

try:
    # 3. 打开连接
    device.open()
    print("连接成功")

    # 4. 收集数据
    facts = device.get_facts()
    print("\n=== 设备基本信息 ===")
    print(f"厂商: {facts.get('vendor')}")
    print(f"型号: {facts.get('model')}")
    print(f"序列号: {facts.get('serial_number')}")
    print(f"OS 版本: {facts.get('os_version')}")
    print(f"主机名: {facts.get('hostname')}")
    print(f"运行时间: {facts.get('uptime')} 秒")
    print(f"接口数: {len(facts.get('interface_list', []))}")

    # 5. 接口信息
    interfaces = device.get_interfaces()
    print("\n=== 接口状态 ===")
    for name, info in interfaces.items():
        status = "UP" if info.get("is_up") else "DOWN"
        speed = info.get("speed", 0)
        print(f"  {name}: {status}, {speed}Mbps")

    # 6. 环境信息
    env = device.get_environment()
    print("\n=== 环境状态 ===")
    for fan_name, fan_info in env.get("fans", {}).items():
        print(f"  风扇 {fan_name}: {fan_info.get('status', 'N/A')}")
    for ps_name, ps_info in env.get("power", {}).items():
        print(f"  电源 {ps_name}: {ps_info.get('status', 'N/A')}")
    for temp_name, temp_info in env.get("temperature", {}).items():
        print(f"  温度 {temp_name}: {temp_info.get('temperature', 'N/A')}°C")

    # 7. LLDP 邻居
    lldp = device.get_lldp_neighbors()
    print("\n=== LLDP 邻居 ===")
    for port, neighbors in lldp.items():
        for neighbor in neighbors:
            print(f"  {port} → {neighbor['hostname']} ({neighbor['port']})")

    # 8. 路由表
    routes = device.get_route_to("0.0.0.0/0")
    print("\n=== 默认路由 ===")
    for route in routes:
        print(f"  下一跳: {route.get('next_hop')}, 出接口: {route.get('interface')}")

finally:
    device.close()

2.2 保存结果为 JSON

#!/usr/bin/env python3
# napalm_collect.py — 批量采集并保存 JSON

from napalm import get_network_driver
import json
import os
from datetime import datetime

DEVICES = {
    "core-sw01": {
        "driver": "huawei_vrp",
        "hostname": "192.168.1.1",
        "username": "admin",
        "password": "admin123",
    },
    "core-router01": {
        "driver": "cisco_xr",
        "hostname": "192.168.2.1",
        "username": "admin",
        "password": "admin123",
    },
}

OUTPUT_DIR = "napalm_collection"

# 要收集的 Getter 方法
GETTERS = [
    "get_facts",
    "get_interfaces",
    "get_interfaces_ip",
    "get_lldp_neighbors",
    "get_bgp_neighbors",
    "get_environment",
    "get_arp_table",
    "get_mac_address_table",
]

os.makedirs(OUTPUT_DIR, exist_ok=True)

for name, dev_info in DEVICES.items():
    print(f"\n=== 正在采集 {name} ===")
    driver = get_network_driver(dev_info["driver"])

    device = driver(
        hostname=dev_info["hostname"],
        username=dev_info["username"],
        password=dev_info["password"],
    )

    try:
        device.open()
        device_data = {}

        for getter in GETTERS:
            try:
                method = getattr(device, getter)
                result = method()
                device_data[getter] = result
                print(f"  ✓ {getter}")
            except NotImplementedError:
                print(f"  ─ {getter}（本设备不支持）")
                device_data[getter] = None
            except Exception as e:
                print(f"  ✗ {getter}: {e}")
                device_data[getter] = str(e)

        # 保存
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filename = os.path.join(OUTPUT_DIR, f"{name}_{timestamp}.json")
        with open(filename, "w", encoding="utf-8") as f:
            json.dump(device_data, f, indent=2, ensure_ascii=False)
        print(f"  已保存到 {filename}")

    except Exception as e:
        print(f"  连接失败: {e}")
    finally:
        device.close()

三、配置管理实战

3.1 配置替换（Replace）

#!/usr/bin/env python3
# napalm_replace.py — 配置替换

from napalm import get_network_driver
from napalm.base.exceptions import (
    ReplaceConfigException,
    MergeConfigException,
)

driver = get_network_driver("huawei_vrp")
device = driver(
    hostname="192.168.1.1",
    username="admin",
    password="admin123",
)

# 新的配置内容（完整替换）
new_config = """
!
interface GigabitEthernet0/0/1
 description UPLINK_TO_CORE
 port link-type trunk
 port trunk allow-pass vlan 10 20 30 40 50
 undo shutdown
!
interface GigabitEthernet0/0/2
 description ACCESS_TO_OFFICE
 port link-type access
 port default vlan 100
 undo shutdown
!
snmp-agent
snmp-agent community read public
snmp-agent sys-info contact IT-Department
!
ntp-service unicast-server 203.0.113.1
!
"""

try:
    device.open()
    print("已连接")

    # 1. 加载候选配置（替换模式）
    print("加载候选配置...")
    device.load_replace_candidate(config=new_config)

    # 2. 比较差异
    diff = device.compare_config()
    if diff:
        print("配置差异：")
        print(diff)

        # 3. 确认提交
        confirm = input("是否提交变更？(yes/no): ")
        if confirm.lower() == "yes":
            device.commit_config()
            print("✓ 配置已提交")
        else:
            device.discard_config()
            print("─ 已丢弃候选配置")
    else:
        print("无配置变更")

except ReplaceConfigException as e:
    print(f"配置替换失败: {e}")
    device.rollback()
finally:
    device.close()

3.2 配置合并（Merge）

#!/usr/bin/env python3
# napalm_merge.py — 配置合并（增量变更）

from napalm import get_network_driver
from napalm.base.exceptions import MergeConfigException

driver = get_network_driver("huawei_vrp")
device = driver(
    hostname="192.168.1.1",
    username="admin",
    password="admin123",
)

# 增量变更：只包含要新增/修改的部分
merge_config = """
!
vlan batch 200 201 202
!
interface Vlanif200
 ip address 10.0.200.1 255.255.255.0
!
interface Vlanif201
 ip address 10.0.201.1 255.255.255.0
!
interface Vlanif202
 ip address 10.0.202.1 255.255.255.0
!
ospf 1
 area 0.0.0.0
  network 10.0.200.0 0.0.0.255
  network 10.0.201.0 0.0.0.255
  network 10.0.202.0 0.0.0.255
!
"""

try:
    device.open()

    # 1. 备份当前配置
    current = device.get_config()
    backup = current.get("running", "")
    print(f"当前配置长度: {len(backup)} 字符")

    # 2. 加载合并配置
    print("加载增量配置...")
    device.load_merge_candidate(config=merge_config)

    # 3. 比较差异
    diff = device.compare_config()
    lines = diff.split("\n")
    added = [l for l in lines if l.startswith("+")]
    removed = [l for l in lines if l.startswith("-")]
    print(f"变更摘要：新增 {len(added)} 行，删除 {len(removed)} 行")
    print("\n详细差异：")
    print(diff)

    # 4. 提交
    device.commit_config()
    print("✓ 增量配置已提交")

except MergeConfigException as e:
    print(f"配置合并失败: {e}")
    device.rollback()
finally:
    device.close()

3.3 配置回滚

#!/usr/bin/env python3
# napalm_rollback.py — 配置回滚

from napalm import get_network_driver
import time

driver = get_network_driver("huawei_vrp")
device = driver(
    hostname="192.168.1.1",
    username="admin",
    password="admin123",
)

# 模拟：变更前保存配置回滚点
try:
    device.open()

    # 保存当前配置到文件（回滚点）
    config = device.get_config()
    with open("rollback_point.cfg", "w") as f:
        f.write(config.get("running", ""))
    print("✓ 已保存回滚点")

    # ... 进行配置变更 ...

    # 如果变更失败，回滚
    print("触发回滚...")
    device.rollback()
    print("✓ 已回滚到上一个配置")

    # 验证
    time.sleep(2)
    current = device.get_config()
    print(f"回滚后配置长度: {len(current.get('running', ''))} 字符")

finally:
    device.close()

四、跨厂商一致性

4.1 同一代码管理多厂商设备

#!/usr/bin/env python3
# napalm_multivendor.py — 跨厂商统一管理

from napalm import get_network_driver
from concurrent.futures import ThreadPoolExecutor, as_completed
import json

DEVICES = [
    {"name": "huawei-core-01", "driver": "huawei_vrp",
     "hostname": "192.168.1.1", "username": "admin", "password": "admin123"},
    {"name": "cisco-core-01", "driver": "ios",
     "hostname": "192.168.2.1", "username": "admin", "password": "admin123"},
    {"name": "juniper-core-01", "driver": "junos",
     "hostname": "192.168.3.1", "username": "admin", "password": "admin123"},
    {"name": "arista-core-01", "driver": "eos",
     "hostname": "192.168.4.1", "username": "admin", "password": "admin123"},
]

def collect_facts(dev_info):
    """设备事实收集（统一 Getter 调用）"""
    driver = get_network_driver(dev_info["driver"])
    device = driver(
        hostname=dev_info["hostname"],
        username=dev_info["username"],
        password=dev_info["password"],
        timeout=30,
    )

    result = {"name": dev_info["name"], "facts": None, "error": None}

    try:
        device.open()
        facts = device.get_facts()       # 统一 API
        interfaces = device.get_interfaces_ip()  # 统一 API
        lldp = device.get_lldp_neighbors()       # 统一 API
        bgp = device.get_bgp_neighbors()         # 统一 API

        result["facts"] = {
            "hostname": facts.get("hostname"),
            "vendor": facts.get("vendor"),
            "model": facts.get("model"),
            "os_version": facts.get("os_version"),
            "uptime": facts.get("uptime"),
            "interface_count": len(facts.get("interface_list", [])),
            "lldp_neighbors": sum(len(n) for n in lldp.values()),
            "bgp_peers": len(bgp.get("global", {}).get("peers", {})),
        }
        print(f"  ✓ {dev_info['name']}: {facts.get('vendor')} {facts.get('model')}")

    except Exception as e:
        result["error"] = str(e)
        print(f"  ✗ {dev_info['name']}: {e}")
    finally:
        device.close()

    return result

print("=== 跨厂商设备信息采集 ===")
results = []

with ThreadPoolExecutor(max_workers=4) as ex:
    futures = [ex.submit(collect_facts, dev) for dev in DEVICES]
    for f in as_completed(futures):
        results.append(f.result())

# 输出汇总
print("\n=== 采集汇总 ===")
print(f"{'设备名':<20} {'厂商':<10} {'型号':<20} {'接口数':<8} {'BGP邻居':<8}")
print("-" * 66)
for r in results:
    if r["facts"]:
        f = r["facts"]
        print(f"{r['name']:<20} {f['vendor']:<10} {f['model']:<20} "
              f"{f['interface_count']:<8} {f['bgp_peers']:<8}")
    else:
        print(f"{r['name']:<20} {'ERROR':<10} {r['error']:<30}")

# 保存汇总
with open("multivendor_summary.json", "w") as f:
    json.dump(results, f, indent=2, ensure_ascii=False)
print("\n汇总已保存到 multivendor_summary.json")

4.2 NAPALM 驱动的关键差异

不同厂商的 NAPALM 支持程度：

注意事项： ┌─ huawei_vrp 需要设备开启 SSH 并支持 display current-configuration ├─ get_lldp_neighbors 在部分交换机支持 ├─ commit_config 在思科 IOS 上需要 archive 配置 └─ rollback 在某些厂商只支持一次回滚

五、NAPALM + Netmiko 组合使用

5.1 取长补短

#!/usr/bin/env python3
# napalm_netmiko_combo.py — 组合使用

from napalm import get_network_driver
from netmiko import ConnectHandler
import json

class HybridManager:
    """NAPALM + Netmiko 混合管理器"""

    def __init__(self, host, username, password, device_type):
        self.host = host
        self.username = username
        self.password = password
        self.device_type = device_type

        # NAPALM 驱动映射
        napalm_drivers = {
            "huawei": "huawei_vrp",
            "cisco_ios": "ios",
            "cisco_xr": "ios_xr",
            "juniper_junos": "junos",
        }
        self.napalm_driver = napalm_drivers.get(device_type, device_type)

    def collect_structured(self):
        """NAPALM：结构化数据收集"""
        driver = get_network_driver(self.napalm_driver)
        device = driver(
            hostname=self.host,
            username=self.username,
            password=self.password,
        )
        data = {}
        try:
            device.open()
            data["facts"] = device.get_facts()
            data["interfaces"] = device.get_interfaces()
            data["lldp"] = device.get_lldp_neighbors()
            data["bgp"] = device.get_bgp_neighbors()
        finally:
            device.close()
        return data

    def execute_custom(self, commands):
        """Netmiko：执行 NAPALM 不支持的 CLI 命令"""
        conn = ConnectHandler(
            device_type=self.device_type,
            host=self.host,
            username=self.username,
            password=self.password,
        )
        results = {}
        try:
            for cmd in commands:
                results[cmd] = conn.send_command(cmd)
        finally:
            conn.disconnect()
        return results

# 使用示例
manager = HybridManager(
    host="192.168.1.1",
    username="admin",
    password="admin123",
    device_type="huawei",
)

# 用 NAPALM 采集结构化数据
structured = manager.collect_structured()
print(f"设备: {structured['facts'].get('hostname')}")
print(f"接口数: {len(structured['interfaces'])}")

# 用 Netmiko 执行自定义命令
custom_commands = [
    "display memory-usage",
    "display cpu-usage",
    "display logbuffer | tail 5",
]
custom = manager.execute_custom(custom_commands)
for cmd, output in custom.items():
    print(f"\n{cmd}:")
    print(output[:200])

六、最佳实践

6.1 安全与可靠操作

NAPALM 生产使用规范：

  1. 配置变更流程
  ┌─ load_replace/merge_candidate → 加载
  ├─ compare_config → 审查差异
  ├─ 人工确认 → commit_config
  └─ 失败 → rollback

  2. 连接管理
  ┌─ 总是用 try/finally 确保 close()
  ├─ 设置合理的 timeout
  ├─ 使用 optional_args 传连接参数
  └─ 避免频繁连接断开（复用连接）

  3. 错误处理
  ┌─ 捕获 ReplaceConfigException
  ├─ 捕获 MergeConfigException
  ├─ 捕获 CommandErrorException
  └─ 记录所有操作到日志

  4. 变更控制
  ┌─ 变更前备份 running-config
  ├─ 变更后验证关键服务
  ├─ 保留回滚点 24-48 小时
  └─ 写入变更审计日志

6.2 NAPALM 适用场景

推荐使用 NAPALM 的场景：

  ┌─ 多厂商统一监控平台
  │  用 get_facts/get_interfaces/get_environment
  │  统一采集所有厂商的设备状态

  ├─ 配置合规检查
  │  用 get_config 获取 running-config
  │  通过 Jinja2 模板对比检查

  ├─ 自动化配置部署
  │  用 load_replace_candidate + commit_config
  │  实现配置的自动化批量部署

  ├─ 网络配置备份
  │  用 get_config 获取 startup/running
  │  定时备份到 Git 仓库

  └─ 变更审计
      用 compare_config 显示变更差异
      用 commit_config 记录变更时间戳

不推荐使用 NAPALM 的场景：

  ┌─ 需要大量自定义 CLI 命令
  │  用 Netmiko 更灵活

  ├─ 仅操作单一厂商
  │  用厂商 SDK 更高效

  └─ 交互式命令
      Paramiko 更可控

七、总结

NAPALM 的核心价值：

  1. 统一抽象
  ┌─ 一套 API 管理所有厂商
  ├─ 相同方法名、相同返回结构
  └─ 厂商差异由 Driver 层屏蔽

  2. 结构化数据
  ┌─ 返回 Python 字典（非文本回显）
  ├─ 可直接用于数据库/监控系统
  └─ 省去正则解析的烦恼

  3. 配置管理标准化
  ┌─ Candidate 模式（先试后提交）
  ├─ diff 能力（变更前审查）
  └─ commit/rollback（安全变更）

  学习路径：
  Paramiko → Netmiko → NAPALM → Ansible
  底层控制    高效操作    统一 API    配置编排

下篇预告：第306篇 — Jinja2 模板引擎：配置模板化，将介绍如何使用 Jinja2 模板将设备配置与数据分离，实现配置的自动化生成与管理。