第328篇：零接触部署与自动上线

关键词

ZTP、零接触部署、自动上线、PXE、DHCP、Auto-Provisioning、初始配置、即插即用

一、ZTP 概述

1.1 什么是零接触部署

传统设备上线流程：

  设备到货 → 工程师现场开箱 → 上架 → 连线
  → 电脑直连 console 口 → 手工输入基础配置
  → 配置管理 IP → 测试连通 → 部署业务配置

  问题：
  ┌─ 每台设备需要工程师现场操作（30-60 分钟/台）
  ├─ 大规模部署时人力投入巨大
  ├─ 人为配置错误（IP 冲突/配置遗漏）
  ├─ 远程站点需派人到现场
  └─ 配置标准化程度低

  ZTP 的解决方案：

  设备上电 → DHCP 获取 IP → 获取配置文件 →
  自动加载配置 → 自动上线 → 加入现网

  优点：
  ┌─ 无需工程师现场操作（即插即用）
  ├─ 10 分钟/台（自动完成）
  ├─ 配置标准化（无人工差异）
  ├─ 远程站点也无需到场
  └─ 配置由服务器统一管控

1.2 ZTP 工作原理

ZTP 工作流程：

  ┌──────────────────────────────────────────┐
  │  新设备上电（出厂状态）                    │
  │  ↓                                         │
  │  1. 设备启动，加载基础软件                │
  │  ↓                                         │
  │  2. DHCP 请求（Option 67/143 等）        │
  │  ├─ 获取 IP 地址                         │
  │  ├─ 获取配置文件服务器地址                │
  │  └─ 获取配置文件路径                      │
  │  ↓                                         │
  │  3. 下载配置文件                          │
  │  ├─ 通过 FTP/TFTP/HTTP/SFTP              │
  │  ├─ 校验配置完整性                        │
  │  └─ 加载配置                              │
  │  ↓                                         │
  │  4. 配置生效                              │
  │  ├─ 应用配置到设备                        │
  │  ├─ 保存配置                              │
  │  └─ 设备重启（可选）                      │
  │  ↓                                         │
  │  5. 自动上线                              │
  │  ├─ 管理通道建立                          │
  │  ├─ 向网管注册                           │
  │  └─ 加入监控和运维体系                    │
  │  ↓                                         │
  │  6. 业务部署                              │
  │  ├─ Ansible/平台下发业务配置              │
  │  ├─ 邻居建立（BGP/OSPF）                │
  │  └─ 正式承载业务                          │
  └──────────────────────────────────────────┘

二、ZTP 基础架构

2.1 硬件架构

ZTP 基础架构部署：

2.2 DHCP 配置

DHCP 服务器配置（以华为 VRP 设备为例）：

  Option 字段说明：
  ┌──────────────────────────────────────────┐
  │  Option 67：配置文件名称（TFTP 方式）    │
  │  Option 143：华为私有，指定配置文件 URL  │
  │  Option 150：TFTP 服务器 IP 地址        │
  │  Option 66：TFTP 服务器名称/IP          │
  │  Option 125：厂商自定义选项              │
  └──────────────────────────────────────────┘

  Linux ISC DHCP 配置示例：
  ┌──────────────────────────────────────────┐
  │  subnet 10.0.0.0 netmask 255.255.255.0 { │
  │    range 10.0.0.100 10.0.0.200;          │
  │    option routers 10.0.0.1;              │
  │    option domain-name-servers 10.0.0.10; │
  │                                           │
  │    # TFTP 服务器地址                      │
  │    option tftp-server-name "10.0.0.10";  │
  │                                           │
  │    # 华为 Option 143（自定义）            │
  │    option space huawei;                   │
  │    option huawei.config-file code 143 =  │
  │      string;                              │
  │    option huawei.config-file              │
  │      "http://10.0.0.10/config/";          │
  │                                           │
  │    # 基于 MAC 指定配置文件               │
  │    host device-1 {                        │
  │      hardware ethernet 00:e0:fc:12:34:56; │
  │      fixed-address 10.0.0.101;           │
  │      option tftp-server-name "10.0.0.10";│
  │      filename "config/device-1.cfg";      │
  │    }                                       │
  │  }                                         │
  └──────────────────────────────────────────┘

三、ZTP 服务器实现

3.1 配置文件管理

#!/usr/bin/env python3
# ztp_server.py — ZTP 配置服务器

from flask import Flask, request, send_file, jsonify
from jinja2 import Environment, FileSystemLoader
import os
import json
import hashlib
import logging
from typing import Dict, Optional

app = Flask(__name__)
logger = logging.getLogger(__name__)

# 配置目录
CONFIG_DIR = "/data/ztp/configs"
TEMPLATE_DIR = "/data/ztp/templates"
DEVICE_REGISTRY = "/data/ztp/device_registry.json"


class ZTPServer:
    """ZTP 配置服务器"""

    def __init__(self):
        self.template_env = Environment(
            loader=FileSystemLoader(TEMPLATE_DIR)
        )
        self._load_registry()

    def _load_registry(self):
        """加载设备注册信息"""
        if os.path.exists(DEVICE_REGISTRY):
            with open(DEVICE_REGISTRY, "r") as f:
                self.registry = json.load(f)
        else:
            self.registry = {}

    def _save_registry(self):
        """保存设备注册信息"""
        with open(DEVICE_REGISTRY, "w") as f:
            json.dump(self.registry, f, indent=2)

    def generate_config(
        self,
        device_mac: str,
        device_type: str = "switch",
        model: str = "",
        **extra_params
    ) -> str:
        """生成设备配置文件"""

        # 构建配置参数
        params = {
            "hostname": f"{device_type}-{device_mac[-8:].replace(':', '')}",
            "management_ip": "",
            "management_mask": "255.255.255.0",
            "management_gateway": "10.0.0.1",
            "dns_servers": ["10.0.0.10"],
            "ntp_servers": ["10.0.0.11"],
            "snmp_community": "public",
            "snmp_location": "Datacenter",
            "device_type": device_type,
            "model": model,
            "domain_name": "example.com",
        }

        # MAC 特定的覆盖参数
        if device_mac in self.registry:
            params.update(self.registry[device_mac])

        # 添加额外参数
        params.update(extra_params)

        # 使用模板生成配置
        template = self.template_env.get_template(
            f"{device_type}_base.j2"
        )
        config = template.render(**params)

        # 保存生成的配置
        config_file = os.path.join(
            CONFIG_DIR, f"{params['hostname']}.cfg"
        )
        with open(config_file, "w") as f:
            f.write(config)

        logger.info(f"为 {device_mac} 生成配置 {params['hostname']}.cfg")
        return config_file

    def register_device(
        self,
        mac: str,
        hostname: str = "",
        role: str = "",
        location: str = "",
        **custom_params
    ):
        """注册设备到 ZTP"""
        if mac not in self.registry:
            self.registry[mac] = {}

        device_info = {
            "hostname": hostname or f"device-{mac[-8:].replace(':', '')}",
            "role": role,
            "location": location,
            "registered_at": __import__(
                "datetime"
            ).datetime.now().isoformat(),
        }
        device_info.update(custom_params)
        self.registry[mac] = device_info
        self._save_registry()

        logger.info(f"设备 {mac} 已注册")
        return device_info

3.2 华为 VRP ZTP 模板

{# huawei_switch_base.j2 — 华为交换机基础配置模板 #}

#
# {{ hostname }} - 自动生成的 ZTP 配置文件
# 生成时间: {{ generated_at }}
#

sysname {{ hostname }}

# 管理接口
interface MEth0/0/1
 ip address {{ management_ip }} {{ management_mask }}
 description Management Interface - ZTP Auto-Provisioning

# 默认路由
ip route-static 0.0.0.0 0.0.0.0 {{ management_gateway }}

# NTP
clock timezone UTC+08:00 08:00
ntp-service unicast-server {{ ntp_servers[0] }}

# DNS
dns resolve
dns server {{ dns_servers[0] }}

# SNMP
snmp-agent
snmp-agent community read {{ snmp_community }}
snmp-agent sys-info location {{ snmp_location }}
snmp-agent sys-info contact {{ snmp_contact | default("noc@example.com") }}

# SSH
stelnet server enable
ssh user admin
ssh user admin authentication-type password
ssh user admin service-type stelnet

# AAA
aaa
 local-user admin password cipher {{ admin_password | default("Admin@123") }}
 local-user admin privilege level 15
 local-user admin service-type ssh

# NMS 注册（向网管系统自动注册）
nms register protocol http server {{ nms_server | default("10.0.0.20") }} port 80

# 保存配置并重启（可选）
save

3.3 Cisco IOS ZTP 模板

! cisco_switch_base.j2 — Cisco IOS 基础配置模板
!
! {{ hostname }} - Auto-generated by ZTP Server
!

hostname {{ hostname }}

! 管理接口
interface GigabitEthernet0/0
 description Management Interface - ZTP
 ip address {{ management_ip }} {{ management_mask }}
 no shutdown
!

! 默认路由
ip route 0.0.0.0 0.0.0.0 {{ management_gateway }}

! NTP
ntp server {{ ntp_servers[0] }}

! DNS
ip name-server {{ dns_servers[0] }}
ip domain-name {{ domain_name }}

! SNMP
snmp-server community {{ snmp_community }} RO
snmp-server location {{ snmp_location }}
snmp-server contact {{ snmp_contact | default("noc@example.com") }}

! SSH
ip ssh version 2
username admin privilege 15 secret {{ admin_password | default("Admin@123") }}
line vty 0 15
 transport input ssh
 login local
!

! 保存配置
write memory

四、完整 ZTP 流程实现

4.1 ZTP 主程序

#!/usr/bin/env python3
# run_ztp.py — ZTP 主程序

from flask import Flask, request, jsonify, send_file
from ztp_server import ZTPServer
import os
import json
import logging

app = Flask(__name__)
ztp = ZTPServer()
logger = logging.getLogger(__name__)


# =============================================
# ZTP API 路由
# =============================================

@app.route("/ztp/v1/register", methods=["POST"])
def register_device():
    """设备/管理员注册新设备到 ZTP"""
    data = request.get_json()
    mac = data.get("mac", "").lower()

    if not mac:
        return jsonify({"error": "MAC 地址不能为空"}), 400

    device_info = ztp.register_device(
        mac=mac,
        hostname=data.get("hostname", ""),
        role=data.get("role", ""),
        location=data.get("location", ""),
        **data.get("params", {}),
    )

    return jsonify({
        "status": "registered",
        "device": device_info,
    })


@app.route("/ztp/v1/config/<device_mac>", methods=["GET"])
def serve_config(device_mac):
    """设备端获取配置文件（设备向 ZTP 服务器请求）"""
    device_mac = device_mac.lower()

    # 检查设备是否已注册
    if device_mac not in ztp.registry:
        # 未注册设备也生成基础配置
        logger.info(f"未注册设备请求配置: {device_mac}")

    # 生成配置
    hostname = ztp.registry.get(
        device_mac, {}
    ).get("hostname", f"device-{device_mac[-8:].replace(':', '')}")

    config_file = ztp.generate_config(
        device_mac=device_mac,
        device_type=request.args.get("type", "switch"),
        generated_at=__import__(
            "datetime"
        ).datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
    )

    return send_file(config_file, as_attachment=True)


@app.route("/ztp/v1/image/<device_mac>", methods=["GET"])
def serve_image(device_mac):
    """设备获取系统软件镜像"""
    device_type = request.args.get("type", "switch")
    model = request.args.get("model", "")

    # 根据型号返回对应镜像
    image_dir = f"/data/ztp/images/{device_type}"
    image_file = f"{model}.bin"

    image_path = os.path.join(image_dir, image_file)
    if os.path.exists(image_path):
        return send_file(image_path, as_attachment=True)

    return jsonify({"error": f"镜像 {image_file} 不存在"}), 404


@app.route("/ztp/v1/status/<device_mac>", methods=["POST"])
def update_status(device_mac):
    """设备上报 ZTP 状态"""
    device_mac = device_mac.lower()
    data = request.get_json()

    status = data.get("status", "")
    ip = data.get("ip", "")
    hostname = data.get("hostname", "")

    # 更新设备状态
    if device_mac in ztp.registry:
        ztp.registry[device_mac].update({
            "ztp_status": status,
            "ztp_ip": ip,
            "hostname": hostname or ztp.registry[device_mac].get("hostname"),
            "last_seen": __import__(
                "datetime"
            ).datetime.now().isoformat(),
        })
        ztp._save_registry()

    logger.info(f"设备 {device_mac} ZTP 状态: {status}")
    return jsonify({"status": "ok"})


@app.route("/ztp/v1/devices", methods=["GET"])
def list_devices():
    """查看所有 ZTP 设备状态"""
    devices = []
    for mac, info in ztp.registry.items():
        devices.append({
            "mac": mac,
            "hostname": info.get("hostname", ""),
            "role": info.get("role", ""),
            "ztp_status": info.get("ztp_status", "pending"),
            "registered_at": info.get("registered_at", ""),
            "last_seen": info.get("last_seen", ""),
        })

    return jsonify({"devices": devices, "total": len(devices)})


# =============================================
# 启动 ZTP 服务器
# =============================================

if __name__ == "__main__":
    # 创建必要的目录
    os.makedirs("/data/ztp/configs", exist_ok=True)
    os.makedirs("/data/ztp/images", exist_ok=True)

    logging.basicConfig(level=logging.INFO)
    logger.info("ZTP 服务器启动中...")

    app.run(host="0.0.0.0", port=8080, debug=True)

4.2 批量注册工具

#!/usr/bin/env python3
# batch_register.py — 批量设备注册

import csv
import json
import requests
import sys


def batch_register_from_csv(csv_file: str, ztp_url: str):
    """从 CSV 批量注册设备到 ZTP 服务器"""

    with open(csv_file, "r") as f:
        reader = csv.DictReader(f)
        devices = list(reader)

    print(f"将注册 {len(devices)} 台设备到 ZTP 服务器...")

    for device in devices:
        payload = {
            "mac": device["mac"].strip().lower(),
            "hostname": device.get("hostname", "").strip(),
            "role": device.get("role", "").strip(),
            "location": device.get("location", "").strip(),
            "params": {
                "management_ip": device.get("mgmt_ip", ""),
                "management_mask": device.get(
                    "mgmt_mask", "255.255.255.0"
                ),
                "management_gateway": device.get(
                    "mgmt_gateway", ""
                ),
                "model": device.get("model", ""),
            },
        }

        try:
            resp = requests.post(
                f"{ztp_url}/ztp/v1/register",
                json=payload,
                timeout=10,
            )
            if resp.status_code == 200:
                result = resp.json()
                print(
                    f"  ✅ {device['mac']} → "
                    f"{result['device']['hostname']}"
                )
            else:
                print(
                    f"  ❌ {device['mac']} → "
                    f"注册失败: {resp.text}"
                )
        except Exception as e:
            print(f"  ❌ {device['mac']} → 请求异常: {e}")

    print("\n批量注册完成！")


# CSV 格式示例：
# mac,hostname,role,location,model,mgmt_ip,mgmt_gateway
# 00:e0:fc:12:34:56,core-sw01,core,DC-A,CE12800,10.0.0.101,10.0.0.1
# 00:e0:fc:78:90:12,acc-sw01,access,DC-A-R01,S5735,10.0.0.102,10.0.0.1

if __name__ == "__main__":
    if len(sys.argv) < 3:
        print("用法: python batch_register.py <csv_file> <ztp_url>")
        print("示例: python batch_register.py devices.csv http://10.0.0.10:8080")
        sys.exit(1)

    batch_register_from_csv(sys.argv[1], sys.argv[2])

五、ZTP 高级功能

5.1 配置版本管理

class ZTPConfigManager:
    """ZTP 配置版本管理"""

    def __init__(self, config_dir: str):
        self.config_dir = config_dir
        self.version_file = os.path.join(
            config_dir, "config_versions.json"
        )
        self._load_versions()

    def _load_versions(self):
        if os.path.exists(self.version_file):
            with open(self.version_file, "r") as f:
                self.versions = json.load(f)
        else:
            self.versions = {}

    def _save_versions(self):
        with open(self.version_file, "w") as f:
            json.dump(self.versions, f, indent=2)

    def create_config_version(
        self, device_mac: str, config_content: str
    ) -> str:
        """创建配置版本"""
        from datetime import datetime
        version_id = datetime.now().strftime(
            "%Y%m%d_%H%M%S"
        )

        if device_mac not in self.versions:
            self.versions[device_mac] = []

        version_info = {
            "version": version_id,
            "content": config_content,
            "created_at": datetime.now().isoformat(),
            "checksum": hashlib.md5(
                config_content.encode()
            ).hexdigest(),
        }

        self.versions[device_mac].append(version_info)

        # 只保留最近 10 个版本
        self.versions[device_mac] = \
            self.versions[device_mac][-10:]

        self._save_versions()
        return version_id

    def rollback_to_version(
        self, device_mac: str, version_id: str
    ) -> Optional[str]:
        """回滚到指定版本"""
        versions = self.versions.get(device_mac, [])
        for v in versions:
            if v["version"] == version_id:
                return v["content"]
        return None

    def get_config_diff(
        self, device_mac: str,
        version_a: str, version_b: str
    ) -> str:
        """比较两个配置版本的差异"""
        import difflib

        config_a = self.rollback_to_version(
            device_mac, version_a
        )
        config_b = self.rollback_to_version(
            device_mac, version_b
        )

        if not config_a or not config_b:
            return "版本不存在"

        diff = difflib.unified_diff(
            config_a.splitlines(keepends=True),
            config_b.splitlines(keepends=True),
            fromfile=version_a,
            tofile=version_b,
        )

        return "".join(diff)

5.2 自动验证

class ZTPVerification:
    """ZTP 自动验证"""

    VERIFICATION_CHECKS = [
        "connectivity",    # 管理 IP 可达
        "snmp",            # SNMP 响应
        "ssh",             # SSH 可登录
        "ntp",             # NTP 同步
        "dns",             # DNS 解析
        "lldp",            # LLDP 邻居发现
    ]

    def verify_device(
        self, ip: str, username: str, password: str
    ) -> Dict:
        """设备上线后自动验证"""
        from netmiko import ConnectHandler

        results = {"ip": ip, "timestamp": None, "checks": {}}

        try:
            conn = ConnectHandler(
                device_type="huawei_vrp",
                host=ip,
                username=username,
                password=password,
            )
            conn.enable()

            # 检查 NTP 同步
            ntp_out = conn.send_command("display ntp status")
            results["checks"]["ntp"] = {
                "passed": "synchronized" in ntp_out.lower(),
                "detail": ntp_out[:100],
            }

            # 检查设备名称
            prompt = conn.find_prompt()
            results["hostname"] = prompt.rstrip("#>")

            # 检查 LLDP
            lldp_out = conn.send_command(
                "display lldp neighbor brief"
            )
            results["checks"]["lldp"] = {
                "passed": bool(lldp_out.strip()),
                "detail": lldp_out[:200],
            }

            conn.disconnect()
            results["timestamp"] = __import__(
                "datetime"
            ).datetime.now().isoformat()
            results["passed"] = all(
                c.get("passed", False)
                for c in results["checks"].values()
            )

        except Exception as e:
            results["passed"] = False
            results["error"] = str(e)

        return results

六、ZTP 部署最佳实践

ZTP 最佳实践总结：

  1. 先规划，再部署
  ┌─ 规划 IP 地址分配（MAC → IP 映射）
  ├─ 准备标准化配置模板
  ├─ 建立设备注册清单（MAC 地址提前录入）
  └─ 测试环境验证 ZTP 流程

  2. 安全加固
  ┌─ DHCP Snooping 防止伪 DHCP 服务器
  ├─ 配置文件签名（防止篡改）
  ├─ 文件服务器访问控制
  └─ 设备上线后修改默认密码

  3. 高可用设计
  ┌─ 主备 ZTP 服务器（VIP 切换）
  ├─ 文件服务器冗余（多副本）
  ├─ DHCP Failover
  └─ 配置模板版本控制（Git）

  4. 流程自动化
  ┌─ 设备上架扫码 → 自动注册 ZTP
  ├─ ZTP 完成 → 自动通知网管
  ├─ 网管验证 → 自动加入 CMDB
  └─ 自动部署业务配置

  5. 监控与告警
  ┌─ 监控 ZTP 服务器健康状态
  ├─ 跟踪 ZTP 成功/失败率
  ├─ 未在规定时间完成 ZTP 的设备告警
  └─ ZTP 失败自动通知工程师

  6. 适用范围
  ┌─ 新数据中心建设：大规模一次性部署
  ├─ 站点扩容：增加少量设备
  ├─ 旧设备替换：更换故障设备
  └─ 远程站点：无需工程师到场

下篇预告：第329篇 — 网络自动化成熟度与演进路径，评估企业网络自动化成熟度，规划从手工到智能的演进路径。

下篇预告：第329篇 — 网络自动化成熟度与演进路径，评估企业网络自动化成熟度，规划从手工到智能的演进路径。