第313篇：用 RESTCONF 修改接口配置

关键词：RESTCONF、接口配置、PUT、PATCH、DELETE、YANG模型、JSON、HTTP方法

一、引言

第311篇介绍了 RESTCONF 作为基于 HTTP 的 YANG 数据操作协议，第312篇演示了 NETCONF 的配置采集。本篇聚焦 RESTCONF 最实用的场景——修改接口配置。

与 NETCONF 使用 XML + RPC 不同，RESTCONF 将 YANG 模型映射为标准的 RESTful API：每个 YANG 节点对应一个 URL 路径，用 HTTP 方法（GET/POST/PUT/PATCH/DELETE）执行 CRUD 操作，数据以 JSON 格式传输。这种设计使得接口配置修改变得直观且易于集成。

二、核心概念

2.1 RESTCONF URL 映射规则

RESTCONF 的核心设计是将 YANG 数据树映射为 URL 路径：

RESTCONF URL 映射规则 - 图示

URL 编码规则：

YANG 元素 URL 表示 示例
模块节点 module-name: ietf-interfaces:
容器 /container/ /interfaces/
列表（单实例） /list= 后跟键值 /interface=GigabitEthernet1%2F0%2F1
列表（多实例） /list= 后跟键名和值 /vlan=vlan-id,100
叶子 /leaf /description
特殊字符 URL 编码 / 编码为 %2F

2.2 HTTP 方法与配置操作的对应

HTTP 方法 配置操作 语义说明 响应码
GET 读取 获取指定资源的当前配置 200 OK
POST 创建 在集合中创建新资源 201 Created
PUT 替换 完整替换指定资源的配置 204 No Content
PATCH 部分更新 仅更新指定的叶子节点 204 No Content
DELETE 删除 删除指定资源 204 No Content

HTTP 方法与配置操作的对应 - 图示

2.3 PUT vs PATCH 的本质区别

操作 行为 示例
PUT 用请求体中的完整配置替换目标资源，未指定的 leaf 被移除或恢复默认 发送 {"name":"GE1/0/1","description":"test"} 会删除该接口的 IP 地址配置
PATCH 仅更新请求体中指定的 leaf，其他 leaf 保持不变 发送 {"description":"test"} 只修改描述，IP 地址不受影响

关键原则：修改单个属性用 PATCH；需要完整替换配置用 PUT。误用 PUT 可能导致配置丢失。

三、技术细节

3.1 修改接口描述和启用状态

# scripts/restconf_modify_interface.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 = {
    "Accept": "application/yang-data+json",
    "Content-Type": "application/yang-data+json",
}
SESSION = requests.Session()
SESSION.auth = AUTH
SESSION.verify = False  # 生产环境使用证书

def get_interface(name):
    """读取接口当前配置"""
    encoded = name.replace("/", "%2F")
    url = f"{BASE_URL}/ietf-interfaces:interfaces/interface={encoded}"
    resp = SESSION.get(url, headers=HEADERS)
    resp.raise_for_status()
    return resp.json()

def update_interface_description(name, description):
    """使用 PATCH 修改接口描述（仅更新 description 字段）"""
    encoded = name.replace("/", "%2F")
    url = f"{BASE_URL}/ietf-interfaces:interfaces/interface={encoded}"

    payload = {
        "ietf-interfaces:interface": {
            "description": description
        }
    }

    resp = SESSION.patch(url, headers=HEADERS, json=payload)
    if resp.status_code in (200, 204):
        print(f"✓ 接口 {name} 描述已更新为: {description}")
        return True
    else:
        print(f"✗ 更新失败: {resp.status_code} - {resp.text}")
        return False

def set_interface_enabled(name, enabled):
    """使用 PATCH 修改接口启用状态"""
    encoded = name.replace("/", "%2F")
    url = f"{BASE_URL}/ietf-interfaces:interfaces/interface={encoded}"

    payload = {
        "ietf-interfaces:interface": {
            "enabled": enabled
        }
    }

    resp = SESSION.patch(url, headers=HEADERS, json=payload)
    if resp.status_code in (200, 204):
        action = "启用" if enabled else "禁用"
        print(f"✓ 接口 {name} 已{action}")
        return True
    else:
        print(f"✗ {action}失败: {resp.status_code} - {resp.text}")
        return False

# 使用示例
interface_name = "GigabitEthernet1/0/1"

# 1. 读取当前配置
print("=== 当前接口配置 ===")
current = get_interface(interface_name)
print(json.dumps(current, indent=2, ensure_ascii=False))

# 2. 修改描述
update_interface_description(interface_name, "Uplink-to-CoreRtr-Modified-by-RESTCONF")

# 3. 禁用接口
set_interface_enabled(interface_name, False)

# 4. 验证修改
print("\n=== 修改后配置 ===")
updated = get_interface(interface_name)
print(json.dumps(updated, indent=2, ensure_ascii=False))

# 5. 恢复启用
set_interface_enabled(interface_name, True)

3.2 修改接口 IP 地址配置

# scripts/restconf_modify_ip.py
import requests
import json

def configure_interface_ip(name, ip_address, prefix_length, description=""):
    """使用 PUT 配置接口的 IP 地址（完整替换 IPv4 配置）"""
    encoded = name.replace("/", "%2F")
    url = f"{BASE_URL}/ietf-interfaces:interfaces/interface={encoded}"

    # 先读取当前配置（保留已有字段）
    current_resp = SESSION.get(url, headers=HEADERS)
    if current_resp.status_code == 200:
        current = current_resp.json()
        interface_data = current.get("ietf-interfaces:interface", {})
    else:
        interface_data = {}

    # 构建新配置（合并到现有配置）
    new_config = {
        "ietf-interfaces:interface": {
            **interface_data,  # 保留现有配置
            "name": name,
            "description": description or interface_data.get("description", ""),
            "type": interface_data.get("type", "iana-if-type:ethernetCsmacd"),
            "enabled": interface_data.get("enabled", True),
            "ietf-ip:ipv4": {
                "address": [
                    {
                        "ip": ip_address,
                        "netmask": f"/{prefix_length}"
                    }
                ]
            }
        }
    }

    resp = SESSION.put(url, headers=HEADERS, json=new_config)
    if resp.status_code in (200, 204):
        print(f"✓ 接口 {name} IP 已配置为 {ip_address}/{prefix_length}")
        return True
    else:
        print(f"✗ IP 配置失败: {resp.status_code} - {resp.text}")
        return False

def remove_interface_ip(name):
    """移除接口的 IP 地址配置"""
    encoded = name.replace("/", "%2F")
    url = f"{BASE_URL}/ietf-interfaces:interfaces/interface={encoded}/ietf-ip:ipv4"

    resp = SESSION.delete(url, headers=HEADERS)
    if resp.status_code in (200, 204):
        print(f"✓ 接口 {name} IP 地址已移除")
        return True
    else:
        print(f"✗ IP 移除失败: {resp.status_code} - {resp.text}")
        return False

# 使用示例
configure_interface_ip(
    "GigabitEthernet1/0/1",
    "10.2.2.1",
    30,
    "Uplink-Link-10.2.2.0/30"
)

3.3 批量修改多接口配置

# scripts/restconf_batch_interfaces.py
import requests
import json
from concurrent.futures import ThreadPoolExecutor, as_completed

# 批量配置模板
BATCH_CONFIG = [
    {
        "interface": "GigabitEthernet1/0/1",
        "description": "To-CoreRtr-A-GigabitEthernet0/0/1",
        "enabled": True,
        "mtu": 9000,
        "ipv4": {"address": "10.2.2.1", "prefix": 30}
    },
    {
        "interface": "GigabitEthernet1/0/2",
        "description": "To-AccessSW-01-GigabitEthernet1/0/1",
        "enabled": True,
        "mtu": 1500,
        "ipv4": {"address": "10.3.3.1", "prefix": 24}
    },
    {
        "interface": "GigabitEthernet1/0/3",
        "description": "To-AccessSW-02-GigabitEthernet1/0/1",
        "enabled": True,
        "mtu": 1500,
        "ipv4": {"address": "10.3.4.1", "prefix": 24}
    },
]

def apply_interface_config(config_item):
    """应用单个接口的配置"""
    encoded = config_item["interface"].replace("/", "%2F")
    url = f"{BASE_URL}/ietf-interfaces:interfaces/interface={encoded}"

    # 构建 PATCH 请求体
    patch_body = {
        "ietf-interfaces:interface": {
            "description": config_item["description"],
            "enabled": config_item["enabled"],
        }
    }

    if "mtu" in config_item:
        patch_body["ietf-interfaces:interface"]["ietf-ip:ipv4"] = {
            "mtu": config_item["mtu"]
        }

    # Step 1: PATCH 描述和状态
    resp = SESSION.patch(url, headers=HEADERS, json=patch_body)
    if resp.status_code not in (200, 204):
        return {
            "interface": config_item["interface"],
            "status": "error",
            "error": f"PATCH failed: {resp.status_code}"
        }

    # Step 2: 配置 IP（如果有）
    if "ipv4" in config_item:
        ip_url = f"{url}/ietf-ip:ipv4/address=inet%3Aipv4-origin%2C{config_item['ipv4']['address']}"

        ip_config = {
            "ietf-ip:ip-address": {
                "ip": config_item["ipv4"]["address"],
                "netmask": f"/{config_item['ipv4']['prefix']}"
            }
        }

        resp = SESSION.put(ip_url, headers=HEADERS, json=ip_config)
        if resp.status_code not in (200, 204):
            return {
                "interface": config_item["interface"],
                "status": "error",
                "error": f"IP config failed: {resp.status_code}"
            }

    return {
        "interface": config_item["interface"],
        "status": "success"
    }

# 并发执行
results = []
with ThreadPoolExecutor(max_workers=3) as executor:
    futures = [
        executor.submit(apply_interface_config, item)
        for item in BATCH_CONFIG
    ]
    for future in as_completed(futures):
        result = future.result()
        results.append(result)
        status = "✓" if result["status"] == "success" else "✗"
        print(f"{status} {result['interface']}: {result['status']}")

# 汇总
success = sum(1 for r in results if r["status"] == "success")
print(f"\n批量配置完成: {success}/{len(results)} 个接口成功")

3.4 创建和删除 VLAN

# scripts/restconf_vlan_operations.py
import requests

def create_vlan(vlan_id, description=""):
    """创建 VLAN"""
    url = f"{BASE_URL}/huawei-vlan:vlan"

    payload = {
        "huawei-vlan:vlan": {
            "vlan-id": vlan_id,
            "description": description
        }
    }

    resp = SESSION.post(url, headers=HEADERS, json=payload)
    if resp.status_code == 201:
        print(f"✓ VLAN {vlan_id} 创建成功")
        return True
    elif resp.status_code == 409:
        print(f"⚠ VLAN {vlan_id} 已存在")
        return False
    else:
        print(f"✗ 创建失败: {resp.status_code} - {resp.text}")
        return False

def get_vlan(vlan_id):
    """读取 VLAN 配置"""
    url = f"{BASE_URL}/huawei-vlan:vlan={vlan_id}"
    resp = SESSION.get(url, headers=HEADERS)
    if resp.status_code == 200:
        return resp.json()
    elif resp.status_code == 404:
        return None
    else:
        raise Exception(f"读取 VLAN 失败: {resp.status_code}")

def delete_vlan(vlan_id):
    """删除 VLAN"""
    url = f"{BASE_URL}/huawei-vlan:vlan={vlan_id}"
    resp = SESSION.delete(url, headers=HEADERS)
    if resp.status_code in (200, 204):
        print(f"✓ VLAN {vlan_id} 已删除")
        return True
    else:
        print(f"✗ 删除失败: {resp.status_code} - {resp.text}")
        return False

# 批量创建 VLAN
vlan_list = [
    (10, "Management"),
    (20, "Servers"),
    (30, "Voice"),
    (100, "Uplink-Trunk"),
    (200, "Server-Network"),
]

for vlan_id, desc in vlan_list:
    create_vlan(vlan_id, desc)

# 验证
for vlan_id, _ in vlan_list:
    vlan = get_vlan(vlan_id)
    if vlan:
        data = vlan.get("huawei-vlan:vlan", {})
        print(f"VLAN {data.get('vlan-id', '?')}: {data.get('description', 'N/A')}")

# 清理：删除测试 VLAN
delete_vlan(200)

3.5 错误处理与批量回滚

# scripts/restconf_error_handling.py
import requests
import json

class RESTCONFError(Exception):
    """RESTCONF 操作异常"""
    def __init__(self, status_code, message, url="", method=""):
        self.status_code = status_code
        self.message = message
        self.url = url
        self.method = method
        super().__init__(f"{method} {url} failed with {status_code}: {message}")

def safe_restconf_request(method, url, payload=None):
    """安全的 RESTCONF 请求，带统一错误处理"""
    headers = HEADERS
    if payload:
        headers["Content-Type"] = "application/yang-data+json"

    resp = SESSION.request(method, url, headers=headers, json=payload)

    # 成功状态码
    if resp.status_code in (200, 201, 204):
        return resp

    # 解析错误信息
    try:
        error_data = resp.json()
        error_msg = error_data.get("errors", {}).get("error", [{}])[0].get("error-message", resp.text)
    except json.JSONDecodeError:
        error_msg = resp.text

    error_map = {
        400: "请求格式错误或参数无效",
        403: "权限不足",
        404: "请求的资源不存在",
        409: "资源冲突（如 VLAN 已存在）",
        412: "前置条件不满足",
        500: "服务器内部错误",
    }

    raise RESTCONFError(resp.status_code, error_msg, url, method)

# 使用示例：批量操作 + 自动回滚
def batch_with_rollback(operations, rollback_on_error=True):
    """批量 RESTCONF 操作，失败时回滚"""
    applied = []

    for op in operations:
        try:
            result = safe_restconf_request(
                op["method"], op["url"], op.get("payload")
            )
            applied.append(op)
            print(f"✓ {op['method']} {op['url']}")
        except RESTCONFError as e:
            print(f"✗ {e}")
            if rollback_on_error and applied:
                print("开始回滚已应用的操作...")
                # 逆序回滚
                for prev_op in reversed(applied):
                    try:
                        if prev_op["method"] == "PATCH":
                            # 对 PATCH 的回滚需要读取旧值并重新 PATCH
                            print(f"  回滚: {prev_op['method']} {prev_op['url']}")
                        elif prev_op["method"] == "POST":
                            safe_restconf_request("DELETE", prev_op["url"])
                            print(f"  已删除: {prev_op['url']}")
                        elif prev_op["method"] == "DELETE":
                            print(f"  恢复: {prev_op['url']} (需手动)")
                    except Exception as rollback_err:
                        print(f"  回滚失败: {rollback_err}")
            break

    return applied

四、实战配置/案例分析

4.1 案例一：接口批量标准化配置

场景：将 50 台交换机的所有 uplink 接口统一配置描述、MTU 和启用状态。

# scripts/interface_standardization.py
import requests
import yaml
import json
from datetime import datetime

def standardize_interfaces(device, uplink_interfaces):
    """标准化配置 uplink 接口"""
    base_url = f"https://{device['ip']}:443/restconf/data"
    auth = (device["username"], device["password"])
    headers = {
        "Accept": "application/yang-data+json",
        "Content-Type": "application/yang-data+json",
    }

    results = []

    for iface in uplink_interfaces:
        encoded = iface.replace("/", "%2F")
        url = f"{base_url}/ietf-interfaces:interfaces/interface={encoded}"

        try:
            session = requests.Session()
            session.auth = auth
            session.verify = False

            # Step 1: 读取当前配置
            resp = session.get(url, headers=headers)
            if resp.status_code == 404:
                results.append({"interface": iface, "status": "not_found"})
                continue

            current = resp.json()
            interface_data = current.get("ietf-interfaces:interface", {})

            # Step 2: 构建标准化配置
            peer_info = iface.get("peer_description", "")
            description = f"To-{device['role']}-{iface['peer_device']}-{peer_info}"

            new_config = {
                "ietf-interfaces:interface": {
                    **interface_data,
                    "name": iface,
                    "description": description,
                    "enabled": True,
                }
            }

            # Step 3: PATCH 更新
            resp = session.patch(url, headers=headers, json=new_config)

            if resp.status_code in (200, 204):
                results.append({
                    "interface": iface,
                    "status": "success",
                    "description": description
                })
            else:
                results.append({
                    "interface": iface,
                    "status": "error",
                    "error": resp.text
                })

        except Exception as e:
            results.append({
                "interface": iface,
                "status": "exception",
                "error": str(e)
            })

    return results

# 设备清单和接口配置
devices = [
    {
        "ip": "10.1.1.1",
        "username": "admin",
        "password": "HuaWei@123",
        "role": "CoreSW",
        "uplink_interfaces": [
            "GigabitEthernet1/0/1",
            "GigabitEthernet1/0/2",
        ]
    },
]

for device in devices:
    print(f"\n配置设备 {device['ip']}...")
    results = standardize_interfaces(device, device["uplink_interfaces"])
    for r in results:
        status = "✓" if r["status"] == "success" else "✗"
        print(f"  {status} {r['interface']}: {r['status']}")

4.2 案例二：故障案例——PUT 导致 IP 配置丢失

故障现象：运维工程师使用 RESTCONF PUT 修改接口描述，结果接口 IP 地址配置丢失，导致网络中断。

故障复现：

# 错误操作：使用 PUT 只发送 description 字段
# PUT /restconf/data/ietf-interfaces:interfaces/interface=GE1%2F0%2F1
wrong_payload = {
    "ietf-interfaces:interface": {
        "description": "Modified-description"
        # 缺少 name, type, enabled, ipv4 等字段
    }
}
# 结果：接口配置被完整替换，IP 地址丢失 → 网络中断

根因分析： - PUT 操作语义是"完整替换"，请求体中未包含的 leaf 会被删除或恢复默认值 - 只发送 description 字段时，ipv4、type、enabled 等字段全部丢失

修复方案：

# 正确操作：方案1 - 使用 PATCH（推荐）
# PATCH /restconf/data/ietf-interfaces:interfaces/interface=GE1%2F0%2F1
correct_payload = {
    "ietf-interfaces:interface": {
        "description": "Modified-description"
    }
}
# 结果：仅更新 description，其他字段不受影响

# 方案2 - 使用 PUT 但包含完整配置
# 先 GET 读取全部配置，修改后再 PUT 回去
current = session.get(url, headers=HEADERS).json()
interface_data = current["ietf-interfaces:interface"]
interface_data["description"] = "Modified-description"
correct_payload = {"ietf-interfaces:interface": interface_data}
session.put(url, headers=HEADERS, json=correct_payload)

预防措施：

预防措施 说明
优先使用 PATCH 修改单个属性时，PATCH 不会删除未指定的字段
PUT 前必须 GET 如果必须用 PUT，先 GET 获取完整配置，修改后再 PUT
变更前备份 使用 GET 保存当前配置快照，出问题时快速回滚
变更窗口 在网络维护窗口内执行 PUT 操作，降低业务影响
自动化测试 在测试环境验证 PUT 请求体包含所有必要字段

4.3 案例三：接口配置模板化批量下发

# scripts/interface_template_deploy.py
import requests
import json
from jinja2 import Template

# Jinja2 接口配置模板
INTERFACE_TEMPLATE = Template("""
{
    "ietf-interfaces:interface": {
        "name": "{{ interface_name }}",
        "description": "{{ description }}",
        "type": "{{ interface_type }}",
        "enabled": {{ enabled|lower }},
        {% if ipv4_address %}
        "ietf-ip:ipv4": {
            "address": [
                {
                    "ip": "{{ ipv4_address }}",
                    "netmask": "/{{ ipv4_prefix }}"
                }
            ]
        }
        {% endif %}
    }
}
""")

def render_interface_config(interface_name, **kwargs):
    """根据模板渲染接口配置"""
    config = INTERFACE_TEMPLATE.render(
        interface_name=interface_name,
        description=kwargs.get("description", ""),
        interface_type=kwargs.get("interface_type", "iana-if-type:ethernetCsmacd"),
        enabled=kwargs.get("enabled", True),
        ipv4_address=kwargs.get("ipv4_address"),
        ipv4_prefix=kwargs.get("ipv4_prefix"),
    )
    return json.loads(config)

# 使用模板批量配置
interfaces_to_config = [
    {
        "name": "GigabitEthernet1/0/1",
        "description": "To-CoreRtr-A-Loopback0",
        "enabled": True,
        "ipv4_address": "10.2.2.1",
        "ipv4_prefix": 30,
    },
    {
        "name": "GigabitEthernet1/0/2",
        "description": "To-AccessSW-01-Uplink",
        "enabled": True,
        "ipv4_address": "10.3.3.1",
        "ipv4_prefix": 24,
    },
    {
        "name": "GigabitEthernet1/0/24",
        "description": "Reserved-Port",
        "enabled": False,
    },
]

for iface_cfg in interfaces_to_config:
    encoded = iface_cfg["name"].replace("/", "%2F")
    url = f"{BASE_URL}/ietf-interfaces:interfaces/interface={encoded}"

    rendered = render_interface_config(iface_cfg["name"], **iface_cfg)

    resp = SESSION.patch(url, headers=HEADERS, json=rendered)
    status = "✓" if resp.status_code in (200, 204) else "✗"
    print(f"{status} {iface_cfg['name']}: {resp.status_code}")

四（续）、高级主题

4.4 RESTCONF 事务与批量操作（yang-patch）

RESTCONF 原生不支持原子事务。当需要"要么全部成功、要么全部回滚"时，可使用 yang-patch（RFC 8072）实现批量变更：

import requests

def apply_yang_patch(session, base_url, patches):
    """
    使用 yang-patch 批量应用配置变更。
    patches 是一个操作列表，支持 add/remove/replace。
    """
    url = f"{base_url}/.yang-patch"
    headers = {
        "Accept": "application/yang-patch+json",
        "Content-Type": "application/yang-patch+json",
    }

    yang_patch = {
        "yang-patch": {
            "yang-patch-operation": patches
        }
    }

    resp = session.post(url, headers=headers, json=yang_patch)
    return resp.status_code in (200, 204)

# 示例：原子性修改两个接口
patches = [
    {
        "operation": "replace",
        "target": "/ietf-interfaces:interfaces/interface=GE1%2F0%2F1/description",
        "source": "To-CoreRtr-A-Uplink"
    },
    {
        "operation": "replace",
        "target": "/ietf-interfaces:interfaces/interface=GE1%2F0%2F2/description",
        "source": "To-CoreRtr-B-Uplink"
    }
]

success = apply_yang_patch(SESSION, BASE_URL, patches)
print(f"{'✓' if success else '✗'} yang-patch 批量操作")

4.5 RESTCONF 配置审计与版本管理

# scripts/restconf_audit.py
import requests
import json
import hashlib
from datetime import datetime

class RESTCONFConfigAuditor:
    """RESTCONF 配置审计器"""

    def __init__(self, base_url, username, password):
        self.session = requests.Session()
        self.session.auth = (username, password)
        self.session.verify = False
        self.headers = {
            "Accept": "application/yang-data+json",
            "Content-Type": "application/yang-data+json",
        }
        self.base_url = base_url

    def capture_snapshot(self, paths=None):
        """采集指定 YANG 路径的配置快照"""
        if paths is None:
            paths = [
                "ietf-interfaces:interfaces",
                "huawei-vlan:vlans",
            ]
        snapshot = {
            "timestamp": datetime.now().isoformat(),
            "hashes": {},
            "configs": {}
        }
        for path in paths:
            encoded_path = path.replace("/", "%2F")
            url = f"{self.base_url}/{encoded_path}"
            resp = self.session.get(url, headers=self.headers)
            if resp.status_code == 200:
                data = resp.json()
                config_str = json.dumps(data, sort_keys=True)
                snapshot["hashes"][path] = hashlib.sha256(config_str.encode()).hexdigest()[:16]
                snapshot["configs"][path] = data
        return snapshot

    def compare_snapshots(self, old, new):
        """对比两个快照，返回变更摘要"""
        changes = []
        for path in old["hashes"]:
            old_hash = old["hashes"][path]
            new_hash = new["hashes"].get(path, "N/A")
            if old_hash != new_hash:
                changes.append({
                    "path": path,
                    "old_hash": old_hash,
                    "new_hash": new_hash,
                    "changed": True
                })
            else:
                changes.append({"path": path, "changed": False})
        return changes

    def save_snapshot(self, snapshot, filename=None):
        """保存快照到文件"""
        if filename is None:
            filename = f"snapshot_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
        with open(filename, "w", encoding="utf-8") as f:
            json.dump(snapshot, f, indent=2, ensure_ascii=False)
        return filename

# 使用示例
auditor = RESTCONFConfigAuditor(
    "https://10.1.1.1:443/restconf/data",
    "admin", "HuaWei@123"
)

# 变更前快照
before = auditor.capture_snapshot()
auditor.save_snapshot(before, "before_change.json")

# ... 执行配置变更 ...

# 变更后快照
after = auditor.capture_snapshot()
auditor.save_snapshot(after, "after_change.json")

# 对比
diff = auditor.compare_snapshots(before, after)
for d in diff:
    status = "变更" if d["changed"] else "未变"
    print(f"{d['path']}: {status} ({d.get('old_hash', '')} → {d.get('new_hash', '')})")

4.6 RESTCONF vs NETCONF 配置下发对比

对比维度 RESTCONF NETCONF
传输协议 HTTP/HTTPS SSH
数据格式 JSON / XML XML
操作语义 RESTful CRUD（GET/PUT/PATCH/POST/DELETE） RPC（get-config/edit-config）
防火墙穿透 简单（443端口） 需开放830端口
事务支持 有限（yang-patch 补充） 原生（candidate + commit + rollback）
流式编辑 不支持 支持（ 流式写入）
API友好度 ⭐⭐⭐（天然 REST） ⭐⭐（XML RPC）
工具生态 curl、Postman、Python requests ncclient、XML 解析

五、总结

操作类型 HTTP 方法 适用场景 注意事项
读取配置 GET 查看当前配置 返回 JSON，可直接解析
修改属性 PATCH 修改描述/状态/MTU 等单个属性 推荐使用，不会删除其他字段
完整替换 PUT 完整配置接口（含 IP 地址） 必须先 GET 读取，修改后再 PUT
创建资源 POST 创建新 VLAN/接口等 在集合节点上使用
删除资源 DELETE 删除 VLAN/IP 地址等 不可逆操作，谨慎使用
最佳实践 说明
修改用 PATCH 日常配置修改优先使用 PATCH，避免 PUT 的"全量替换"风险
PUT 前备份 使用 PUT 时，先 GET 保存当前配置快照
URL 编码 接口名中的 / 必须编码为 %2F
错误码处理 404 表示资源不存在，409 表示冲突，500 表示服务器错误
批量操作顺序 先创建依赖资源（如 VLAN），再配置引用它的资源（如接口）
会话复用 使用 requests.Session() 复用连接，减少 TLS 握手开销

思考题

  1. 使用 RESTCONF 修改接口 IP 地址时，为什么推荐用 PATCH 而不是 PUT？如果用 PUT，需要什么前置步骤？
  2. RESTCONF 返回 409 Conflict 状态码通常表示什么？如何避免？
  3. 如果一次 PATCH 请求同时修改 description 和 enabled 两个字段，这两个修改是原子操作吗？

下篇预告：第314篇：OpenConfig 与 IETF 标准模型