第311篇：NETCONF 与 RESTCONF 协议

关键词：NETCONF、RESTCONF、RPC、YANG模型、XML、JSON、配置管理协议、原子事务

一、引言

第310篇介绍了 YANG 数据模型如何为网络配置提供结构化描述。然而，仅有数据模型还不够——我们需要一种协议来传输这些结构化数据、执行读写操作、并保证操作的原子性与一致性。这就是 NETCONF 和 RESTCONF 的使命。

NETCONF（Network Configuration Protocol，RFC 6241）和 RESTCONF（RFC 8040）是模型驱动网络管理的两大传输协议。它们以 YANG 模型为操作蓝本，将设备配置和状态数据抽象为可编程的数据节点，彻底改变了传统的 SSH + CLI 管理模式。

二、核心概念

2.1 NETCONF 协议体系

NETCONF 是一个基于 SSH 传输层的配置管理协议，其核心设计理念是：将设备配置视为数据库记录，通过标准的 RPC（Remote Procedure Call）接口进行 CRUD 操作。

NETCONF 消息由三部分组成：Hello 协商、RPC 请求/回复、通知（Notification）。所有消息以 <hello> 开始，以 ]]>]]> 结束（SSH 传输层使用此边界标记）。

2.2 NETCONF vs RESTCONF 对比

2.3 NETCONF 数据区模型

NETCONF 定义了三个逻辑数据区，实现配置变更的"编辑-验证-提交"工作流：

注意：如果设备不支持 candidate capability，则只能直接操作 running 数据区，此时 edit-config 操作直接生效，无候选机制。

2.4 核心 RPC 操作一览

三、技术细节

3.1 NETCONF 能力协商（Hello 消息）

NETCONF 会话建立时，客户端和服务端交换 capabilities，决定本次会话支持的功能：

<!-- 客户端发送的 Hello 消息 -->
<hello xmlns="urn:ietf:params:xml:ns:netconf:base:1.0">
  <capabilities>
    <capability>urn:ietf:params:netconf:base:1.1</capability>
    <capability>urn:ietf:params:netconf:capability:candidate:1.1</capability>
    <capability>urn:ietf:params:netconf:capability:rollback-on-error:1.1</capability>
    <capability>urn:ietf:params:netconf:capability:validate:1.1</capability>
    <capability>urn:ietf:params:netconf:capability:url:1.0?scheme=file,tftp,ftp,https</capability>
  </capabilities>
</hello>
]]>]]>

<!-- 服务端返回的 Hello 消息 -->
<hello xmlns="urn:ietf:params:xml:ns:netconf:base:1.0">
  <capabilities>
    <capability>urn:ietf:params:netconf:base:1.1</capability>
    <capability>urn:ietf:params:netconf:capability:candidate:1.1</capability>
    <capability>urn:ietf:params:netconf:capability:validate:1.1</capability>
    <capability>urn:ietf:params:netconf:capability:rollback-on-error:1.1</capability>
    <capability>urn:ietf:params:netconf:capability:notification:1.0</capability>
    <capability>urn:huawei:params:netconf:capability:device-params:1.0</capability>
  </capabilities>
</hello>
]]>]]>

常用 capabilities 说明：

3.2 NETCONF 实战：Python ncclient 库

$ pip install ncclient lxml

3.2.1 基础连接与读取

# scripts/netconf_basic.py
from ncclient import manager
import xml.etree.ElementTree as ET

# 华为设备 NETCONF 连接参数
DEVICE = {
    "host": "10.1.1.1",
    "port": 830,
    "username": "admin",
    "password": "HuaWei@123",
    "device_params": {"name": "huawei"},
    "hostkey_verify": False,
    "timeout": 30,
}

def get_running_interfaces(m):
    """获取运行配置中的接口信息"""
    filter_xml = """
    <filter type="subtree">
        <interfaces xmlns="urn:ietf:params:xml:ns:yang:ietf-interfaces">
            <interface>
                <name>GigabitEthernet1/0/1</name>
            </interface>
        </interfaces>
    </filter>
    """
    reply = m.get_config(source="running", filter=filter_xml)
    root = ET.fromstring(ET.tostring(reply.data_element))

    for elem in root.iter():
        tag = elem.tag.split("}")[-1]  # 去除命名空间
        if elem.text and elem.text.strip():
            print(f"  {tag}: {elem.text.strip()}")

# 执行
with manager.connect(**DEVICE) as m:
    print(f"=== 设备 Capabilities ===")
    for cap in m.server_capabilities:
        print(f"  {cap}")
    print(f"\n=== 接口 GigabitEthernet1/0/1 ===")
    get_running_interfaces(m)

3.2.2 原子事务：Lock → Edit → Validate → Commit

# scripts/netconf_transaction.py
from ncclient import manager
from ncclient.operations import RPCError

def atomic_config_change(m, config_xml):
    """原子配置变更：锁定 → 编辑 → 验证 → 提交 → 解锁"""
    try:
        # Step 1: 锁定候选配置区（排他锁，防止并发修改）
        print("Step 1: 锁定候选配置区...")
        m.lock("candidate")

        # Step 2: 下发配置到候选区
        print("Step 2: 下发配置到候选区...")
        m.edit_config(target="candidate", config=config_xml)

        # Step 3: 验证配置合法性
        print("Step 3: 验证候选配置...")
        try:
            m.validate(source="candidate")
            print("  ✓ 配置验证通过")
        except RPCError as e:
            print(f"  ✗ 配置验证失败: {e.message}")
            raise

        # Step 4: 提交配置
        print("Step 4: 提交配置到运行区...")
        m.commit()
        print("  ✓ 配置提交成功")

    except Exception as e:
        print(f"配置变更失败: {e}")
        # 回滚：丢弃候选区中的未提交变更
        try:
            m.discard_changes()
            print("  已回滚候选区变更")
        except Exception:
            pass
    finally:
        # 确保解锁（无论成功或失败）
        try:
            m.unlock("candidate")
            print("Step 5: 已解锁候选配置区")
        except Exception:
            pass

# 配置变更示例：修改接口描述 + 创建 VLAN
config_xml = """
<config>
    <interfaces xmlns="urn:ietf:params:xml:ns:yang:ietf-interfaces">
        <interface>
            <name>GigabitEthernet1/0/1</name>
            <description>Uplink-to-CoreRtr-Modified</description>
        </interface>
    </interfaces>
    <vlans xmlns="urn:huawei:params:xml:ns:yang:huawei-vlan">
        <vlan>
            <vlan-id>200</vlan-id>
            <description>Server-Network</description>
        </vlan>
    </vlans>
</config>
"""

with manager.connect(**DEVICE) as m:
    atomic_config_change(m, config_xml)

3.2.3 edit-config 的 operation 属性

edit-config 支持四种操作模式，决定如何处理目标节点：

# scripts/netconf_edit_operations.py
from ncclient import manager

with manager.connect(**DEVICE) as m:

    # ========== merge: 合并模式（默认） ==========
    # 只更新指定的 leaf，不影响其他 leaf
    merge_config = """
    <config>
        <interfaces xmlns="urn:ietf:params:xml:ns:yang:ietf-interfaces">
            <interface>
                <name>GigabitEthernet1/0/1</name>
                <description>Merge-mode</description>
            </interface>
        </interfaces>
    </config>
    """
    m.edit_config(target="candidate", config=merge_config)
    # 结果：只修改 description，其他配置（如 IP 地址）保持不变

    # ========== replace: 替换模式 ==========
    # 用新数据完整替换目标节点，未指定的 leaf 会被删除
    replace_config = """
    <config>
        <interfaces xmlns="urn:ietf:params:xml:ns:yang:ietf-interfaces">
            <interface xmlns:nc="urn:ietf:params:xml:ns:netconf:base:1.0" nc:operation="replace">
                <name>GigabitEthernet1/0/1</name>
                <description>Replace-mode</description>
                <enabled>true</enabled>
            </interface>
        </interfaces>
    </config>
    """
    # 结果：GigabitEthernet1/0/1 的配置被完整替换

    # ========== create: 创建模式 ==========
    # 仅当节点不存在时创建，已存在则报错
    create_config = """
    <config>
        <vlans xmlns="urn:huawei:params:xml:ns:yang:huawei-vlan">
            <vlan xmlns:nc="urn:ietf:params:xml:ns:netconf:base:1.0" nc:operation="create">
                <vlan-id>300</vlan-id>
            </vlan>
        </vlans>
    </config>
    """
    m.edit_config(target="candidate", config=create_config)

    # ========== remove: 删除模式 ==========
    # 删除指定的节点
    remove_config = """
    <config>
        <vlans xmlns="urn:huawei:params:xml:ns:yang:huawei-vlan">
            <vlan xmlns:nc="urn:ietf:params:xml:ns:netconf:base:1.0" nc:operation="remove">
                <vlan-id>300</vlan-id>
            </vlan>
        </vlans>
    </config>
    """
    m.edit_config(target="candidate", config=remove_config)

3.3 RESTCONF 实战：Python requests 库

RESTCONF 将 YANG 模型映射为 RESTful API，URL 路径直接对应 YANG 树节点：

3.3.1 RESTCONF 基本操作

# scripts/restconf_basic.py
import requests
import json
from requests.auth import HTTPBasicAuth

BASE_URL = "https://10.1.1.1:443/restconf/data"
AUTH = HTTPBasicAuth("admin", "HuaWei@123")
HEADERS_JSON = {
    "Accept": "application/yang-data+json",
    "Content-Type": "application/yang-data+json",
}
HEADERS_XML = {
    "Accept": "application/yang-data+xml",
    "Content-Type": "application/yang-data+xml",
}
SESSION = requests.Session()
SESSION.auth = AUTH
SESSION.verify = False  # 生产环境应使用证书验证

# ========== GET: 读取全部接口 ==========
def get_all_interfaces():
    resp = SESSION.get(
        f"{BASE_URL}/ietf-interfaces:interfaces",
        headers=HEADERS_JSON
    )
    resp.raise_for_status()
    data = resp.json()
    return data.get("ietf-interfaces:interfaces", {}).get("interface", [])

# ========== GET: 读取指定接口 ==========
def get_interface(name):
    encoded_name = name.replace("/", "%2F")
    resp = SESSION.get(
        f"{BASE_URL}/ietf-interfaces:interfaces/interface={encoded_name}",
        headers=HEADERS_JSON
    )
    resp.raise_for_status()
    return resp.json()

# ========== PUT: 完整替换接口配置 ==========
def replace_interface(name, config):
    encoded_name = name.replace("/", "%2F")
    resp = SESSION.put(
        f"{BASE_URL}/ietf-interfaces:interfaces/interface={encoded_name}",
        headers=HEADERS_JSON,
        json=config
    )
    # 204 No Content = 成功
    return resp.status_code in (200, 204)

# ========== PATCH: 部分更新 ==========
def patch_interface(name, updates):
    encoded_name = name.replace("/", "%2F")
    resp = SESSION.patch(
        f"{BASE_URL}/ietf-interfaces:interfaces/interface={encoded_name}",
        headers=HEADERS_JSON,
        json=updates
    )
    return resp.status_code in (200, 204)

# ========== POST: 创建新 VLAN ==========
def create_vlan(vlan_id, description=""):
    resp = SESSION.post(
        f"{BASE_URL}/huawei-vlan:vlan",
        headers=HEADERS_JSON,
        json={
            "huawei-vlan:vlan": {
                "vlan-id": vlan_id,
                "description": description
            }
        }
    )
    # 201 Created = 成功
    return resp.status_code == 201

# ========== DELETE: 删除资源 ==========
def delete_vlan(vlan_id):
    resp = SESSION.delete(
        f"{BASE_URL}/huawei-vlan:vlan={vlan_id}",
        headers=HEADERS_JSON
    )
    return resp.status_code in (200, 204)

# 使用示例
if __name__ == "__main__":
    # 读取接口
    interfaces = get_all_interfaces()
    print(f"设备共有 {len(interfaces)} 个接口")

    # 修改接口描述
    patch_interface("GigabitEthernet1/0/1", {
        "ietf-interfaces:interface": {
            "description": "Updated-via-RESTCONF-PATCH"
        }
    })

    # 创建 VLAN
    create_vlan(200, "Server-Network-RESTCONF")

3.3.2 RESTCONF 状态数据读取

RESTCONF 不仅可以读取配置数据（config=true），还可以读取状态数据（config=false），如接口流量统计：

# scripts/restconf_state_data.py
import requests

# 读取接口状态（包含 traffic statistics）
resp = SESSION.get(
    f"{BASE_URL}/ietf-interfaces:interfaces-state",
    headers=HEADERS_JSON
)

stats = resp.json()
for iface in stats.get("ietf-interfaces:interfaces-state", {}).get("interface", []):
    name = iface.get("name", "?")
    oper_status = iface.get("oper-status", "?")
    stats_inner = iface.get("statistics", {})
    in_octets = stats_inner.get("in-octets", 0)
    out_octets = stats_inner.get("out-octets", 0)

    print(f"接口 {name}:")
    print(f"  状态: {oper_status}")
    print(f"  入向流量: {in_octets:,} bytes")
    print(f"  出向流量: {out_octets:,} bytes")

3.4 NETCONF 事件通知订阅

NETCONF 的 <subscribe> 机制允许管理站实时接收设备的事件通知（如配置变更、接口 Down/Up）：

# scripts/netconf_subscribe.py
from ncclient import manager
import xml.etree.ElementTree as ET

with manager.connect(**DEVICE) as m:

    # ========== 方式一：持续订阅（Streaming） ==========
    # 订阅配置变更通知
    filter_xml = """
    <filter type="subtree">
        <EVENTNOTIF xmlns="urn:ietf:params:xml:ns:yang:ietf-event-notifications">
            <config-changes>
                <config-change>
                    <change>
                        <changed-value>
                            <interfaces xmlns="urn:ietf:params:xml:ns:yang:ietf-interfaces"/>
                        </changed-value>
                    </change>
                </config-change>
            </config-changes>
        </EVENTNOTIF>
    </filter>
    """

    subscription = m.create_subscription(filter=filter_xml)

    print("开始监听配置变更通知...\n")
    try:
        for notification in subscription.listen(timeout=5):
            print(f"收到通知:")
            print(ET.tostring(notification, pretty_print=True).decode())
            print("---")
    except KeyboardInterrupt:
        print("\n停止监听")
        subscription.stop()

四、实战配置/案例分析

4.1 案例一：多设备批量配置下发

场景：在 100 台交换机上统一创建 VLAN 并配置接口描述。

# scripts/batch_netconf_deploy.py
from ncclient import manager
from ncclient.operations import RPCError
from concurrent.futures import ThreadPoolExecutor, as_completed
import yaml
import logging

logging.basicConfig(level=logging.INFO,
                    format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)

# 从 YAML 读取设备清单
with open("devices/inventory.yaml", "r", encoding="utf-8") as f:
    inventory = yaml.safe_load(f)

devices = inventory.get("devices", [])

def deploy_vlan_config(device, vlan_id, vlan_name, interfaces):
    """在单台设备上批量配置 VLAN 和接口"""
    # 构建批量配置 XML
    vlan_xml = f"""
    <config>
        <vlans xmlns="urn:huawei:params:xml:ns:yang:huawei-vlan">
            <vlan>
                <vlan-id>{vlan_id}</vlan-id>
                <description>{vlan_name}</description>
            </vlan>
        </vlans>
    """

    for iface_name in interfaces:
        vlan_xml += f"""
        <interfaces xmlns="urn:ietf:params:xml:ns:yang:ietf-interfaces">
            <interface>
                <name>{iface_name}</name>
                <description>{vlan_name}-Port</description>
            </interface>
        </interfaces>
        """
    vlan_xml += "</config>"

    try:
        with manager.connect(
            host=device["host"],
            port=830,
            username=device["username"],
            password=device["password"],
            hostkey_verify=False,
            device_params={"name": "huawei"},
            timeout=30
        ) as m:
            m.lock("candidate")
            m.edit_config(target="candidate", config=vlan_xml)
            m.validate(source="candidate")
            m.commit()
            m.unlock("candidate")
            return {"host": device["host"], "status": "success"}

    except RPCError as e:
        logger.error(f"{device['host']} RPC 错误: {e.message}")
        return {"host": device["host"], "status": "rpc_error", "error": str(e)}
    except Exception as e:
        logger.error(f"{device['host']} 异常: {e}")
        return {"host": device["host"], "status": "error", "error": str(e)}

# 并发执行（最多 5 个并发）
results = []
with ThreadPoolExecutor(max_workers=5) as executor:
    futures = {
        executor.submit(deploy_vlan_config, dev, 100, "Management-VLAN",
                       ["GigabitEthernet1/0/1"]): dev
        for dev in devices
    }
    for future in as_completed(futures):
        result = future.result()
        results.append(result)
        status = "✓" if result["status"] == "success" else "✗"
        logger.info(f"{status} {result['host']}: {result['status']}")

# 汇总报告
success_count = sum(1 for r in results if r["status"] == "success")
logger.info(f"部署完成: {success_count}/{len(results)} 台设备成功")

4.2 案例二：RESTCONF 配置一致性核查

场景：定期检查 50 台设备的接口描述是否遵循命名规范。

# scripts/restconf_compliance_check.py
import requests
import json
from datetime import datetime

BASE_URL = "https://{ip}:443/restconf/data"
HEADERS = {
    "Accept": "application/yang-data+json",
    "Content-Type": "application/yang-data+json",
}

# 合规规则：接口描述必须包含 "To-" 或 "Uplink" 或 "Access"
DESCRIPTION_PATTERN = r"^(To-|Uplink|Access).+"

def check_device_compliance(device):
    """检查单台设备的接口描述合规性"""
    url = BASE_URL.format(ip=device["ip"]) + "/ietf-interfaces:interfaces"
    auth = (device["username"], device["password"])

    try:
        resp = requests.get(url, auth=auth, headers=HEADERS,
                           verify=False, timeout=10)
        resp.raise_for_status()
    except requests.RequestException as e:
        return {"host": device["ip"], "status": "unreachable", "error": str(e)}

    data = resp.json()
    interfaces = data.get("ietf-interfaces:interfaces", {}).get("interface", [])

    non_compliant = []
    for iface in interfaces:
        name = iface.get("name", "")
        desc = iface.get("description", "")

        # 仅检查物理接口
        if "GigabitEthernet" not in name and "TenGigabitEthernet" not in name:
            continue

        import re
        if not re.match(DESCRIPTION_PATTERN, desc):
            non_compliant.append({
                "interface": name,
                "description": desc,
                "issue": "描述不符合命名规范"
            })

    compliant = len(interfaces) - len(non_compliant)
    return {
        "host": device["ip"],
        "status": "success",
        "total_interfaces": len(interfaces),
        "compliant": compliant,
        "non_compliant": non_compliant
    }

# 批量检查
devices = [
    {"ip": "10.1.1.1", "username": "admin", "password": "HuaWei@123"},
    {"ip": "10.1.1.2", "username": "admin", "password": "HuaWei@123"},
]

report = {
    "timestamp": datetime.now().isoformat(),
    "results": []
}

for device in devices:
    result = check_device_compliance(device)
    report["results"].append(result)

# 输出报告
with open("compliance_report.json", "w", encoding="utf-8") as f:
    json.dump(report, f, indent=2, ensure_ascii=False)

# 汇总
for r in report["results"]:
    if r["status"] == "success" and r["non_compliant"]:
        print(f"\n{r['host']} 存在 {len(r['non_compliant'])} 个不合规接口:")
        for nc in r["non_compliant"]:
            print(f"  {nc['interface']}: \"{nc['description']}\"")

4.3 案例三：NETCONF 配置回滚机制

场景：变更后自动验证连通性，失败则回滚。

# scripts/netconf_rollback.py
from ncclient import manager
import subprocess
import time

def apply_with_rollback(m, config_xml, verify_command, timeout=30):
    """应用配置并验证，失败则回滚"""
    try:
        # 1. 备份当前配置
        backup = m.get_config(source="running",
            filter=('subtree', '<config/>'))

        # 2. 下发配置
        m.edit_config(target="candidate", config=config_xml)
        m.commit()

        print("配置已应用，等待验证...")
        time.sleep(5)  # 等待配置生效

        # 3. 验证连通性
        result = subprocess.run(
            verify_command,
            shell=True,
            capture_output=True,
            text=True,
            timeout=timeout
        )

        if result.returncode == 0:
            print("✓ 验证通过，配置生效")
            return True
        else:
            print("✗ 验证失败，开始回滚...")
            raise Exception(f"验证失败: {result.stderr}")

    except Exception as e:
        print(f"错误: {e}")
        try:
            # 回滚：用备份的配置覆盖当前配置
            backup_config = ET.tostring(
                backup.data_element, pretty_print=True
            ).decode()
            m.edit_config(target="running", config=backup_config)
            print("✓ 已回滚到变更前配置")
        except Exception as rollback_err:
            print(f"✗ 回滚也失败了: {rollback_err}")
        return False

# 使用示例
config_xml = """
<config>
    <interfaces xmlns="urn:ietf:params:xml:ns:yang:ietf-interfaces">
        <interface>
            <name>GigabitEthernet1/0/1</name>
            <ipv4 xmlns="urn:ietf:params:xml:ns:yang:ietf-ip">
                <address>
                    <ip>10.2.2.1</ip>
                    <netmask>255.255.255.0</netmask>
                </address>
            </ipv4>
        </interface>
    </interfaces>
</config>
"""

verify_cmd = "ping -c 3 -W 2 10.2.2.2"

with manager.connect(**DEVICE) as m:
    success = apply_with_rollback(m, config_xml, verify_cmd)

五、总结

思考题

1. NETCONF 的 candidate 数据区为什么能实现原子提交？如果 commit 过程中设备断电，配置状态会如何？
2. RESTCONF 的 PUT 和 PATCH 在语义上有什么本质区别？修改接口描述应该用哪个？为什么？
3. NETCONF 的 <edit-config> 中 merge 和 replace 操作在批量配置下发时如何选择？用错了会有什么后果？

下篇预告：第312篇：用 NETCONF 采集设备配置