第348篇:全网 IP 地址梳理与重规划项目

关键词:IP地址规划、地址重规划、CIDR汇总、IPv6迁移、自动化脚本


一、案例背景

某大型企业集团经过20年的发展,网络规模从最初的100人扩展到现在的15000人,拥有30个办公地点、5个数据中心。由于早期缺乏统一的IP地址规划,现网IP地址存在以下问题:

  1. 地址碎片化严重:大量/28、/29小网段散布全网,无法有效汇总
  2. 地址重叠:不同站点的子网存在重叠(如3个站点都使用192.168.1.0/24)
  3. 地址浪费:部分站点分配了/22但实际只使用了50个地址
  4. 路由表膨胀:核心设备路由表超过10000条(其中80%是明细路由)
  5. 无地址文档:没有完整的IP地址分配记录

二、需求分析与方案设计

2.1 IP地址审计

问题类型 具体表现 影响
碎片化 大量不连续的小网段 路由无法汇总
重叠 不同站点子网重叠 VPN无法互通
浪费 大网段低利用率 地址资源紧张
无文档 无IP地址分配记录 故障排查困难
规划缺失 无IPv6规划 无法满足监管要求

2.2 重规划架构

重规划架构 - 图示

2.3 新地址规划原则

原则 说明 示例
层次化 按公司→站点→楼层→VLAN层次分配 10.X.Y.Z
可汇总 相邻网段必须可汇总为更大CIDR 10.10.0.0/16
可扩展 每个层级预留扩展空间 预留50%地址
无重叠 全网地址唯一 全局唯一分配
标准化 按用途分配固定地址段 服务器10.100.x.x

2.4 地址编码规则详解

新地址规划采用三层编码规则:10.[DC].[功能].[设备序号].0/24

第二段(DC编码):

DC编号 站点 汇总路由
1 北京总部 10.1.0.0/16
2 上海分公司 10.2.0.0/16
3 深圳分公司 10.3.0.0/16
100-110 数据中心 10.100.0.0/13
200-210 管理网络 10.200.0.0/14

第三段(功能编码):

功能码 用途 示例
0 网络设备管理 10.1.0.x
1-10 接入VLAN(办公网) 10.1.1.x(1楼)
11-20 接入VLAN(服务器) 10.100.11.x(Web)
21-30 核心业务 10.100.21.x(数据库)
50-60 存储网络 10.201.50.x
80-90 VIP/负载均衡 10.100.80.x
250-254 预留扩展 -

编码规则的优势:通过地址即可判断设备归属(哪个DC、什么功能、哪台设备),大幅提升故障排查效率。

三、技术细节与实施过程

3.1 现状审计(自动化采集)

# ip_audit.py - 全网IP地址审计脚本
import paramiko
import csv
from collections import defaultdict

devices = [
    {"host": "10.0.0.1", "user": "admin", "password": "***", "name": "Core-01"},
    # ... 更多设备
]

ip_usage = defaultdict(lambda: {"device": "", "interface": "", "subnet": "", "assigned": 0})

for device in devices:
    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    ssh.connect(device["host"], username=device["user"], password=device["password"])

    # 采集IP地址配置
    stdin, stdout, stderr = ssh.exec_command("display ip interface brief")
    output = stdout.read().decode()

    for line in output.split("\n"):
        if "/" in line and "up" in line.lower():
            parts = line.split()
            interface = parts[0]
            ip_addr = parts[1]
            mask = parts[2] if len(parts) > 2 else "255.255.255.0"
            ip_usage[ip_addr]["device"] = device["name"]
            ip_usage[ip_addr]["interface"] = interface
            ip_usage[ip_addr]["subnet"] = f"{ip_addr}/{mask}"

    ssh.close()

# 输出审计结果
with open("ip_audit_result.csv", "w", newline="") as f:
    writer = csv.DictWriter(f, fieldnames=["IP", "Device", "Interface", "Subnet"])
    writer.writeheader()
    for ip, info in ip_usage.items():
        writer.writerow({"IP": ip, **info})

print(f"Audit complete: {len(ip_usage)} IP addresses found")

3.2 新地址方案

层级 地址段 CIDR 用途 容量
总部办公 10.10.0.0 /16 北京总部 65534
上海办公 10.20.0.0 /16 上海分公司 65534
深圳办公 10.30.0.0 /16 深圳分公司 65534
数据中心A 10.100.0.0 /16 北京DC 65534
数据中心B 10.101.0.0 /16 上海DC 65534
管理网络 10.200.0.0 /16 全网设备管理 65534
存储网络 10.201.0.0 /16 存储互联 65534

3.3 迁移策略

迁移方式 适用场景 风险等级
双栈并行 核心业务、数据库 低
NAT转换 遗留系统、第三方设备 中
直接替换 新建业务、可停机系统 中
DHCP重新分配 终端接入网络 低

四、实战配置/案例分析

4.1 地址汇总优化

优化前(路由表片段):

10.1.1.0/24    via 10.0.0.1
10.1.2.0/24    via 10.0.0.1
10.1.3.0/24    via 10.0.0.1
10.1.4.0/24    via 10.0.0.1
# ... 64条明细路由(总部64个部门)

优化后(汇总路由):

10.1.0.0/16    via 10.0.0.1    # 总部汇总路由(仅1条)
# 在总部汇聚层配置汇总路由
[Aggregation-SW] ip route-static 10.1.0.0 255.255.0.0 10.1.1.1 preference 60

# 在接入层配置明细路由指向汇聚层
[Access-SW-Dept01] ip route-static 0.0.0.0 0.0.0.0 10.1.1.254

4.2 地址迁移实施

# 方案:双栈并行(以服务器迁移为例)

# 步骤1:服务器配置新IP(保持旧IP)
# 在服务器上添加新IP地址
ifconfig eth0:1 10.100.1.10 netmask 255.255.255.0 up

# 步骤2:应用层逐步切换
# 修改应用配置,优先使用新IP
# 监控新IP的流量,确认业务正常

# 步骤3:网络层切换
# 在交换机上更新DHCP/静态绑定
[Access-SW] vlan 100
[Access-SW-vlan100] dhcp server ip-pool SERVER-POOL
[Access-SW-vlan100-dhcp-SERVER-POOL] gateway-list 10.100.1.1
[Access-SW-vlan100-dhcp-SERVER-POOL] section 10.100.1.100 10.100.1.200

# 步骤4:回收旧地址
# 确认新地址运行稳定后(至少1周),删除旧IP
ifconfig eth0:0 down

4.3 IPAM系统部署

# 使用NetBox作为IPAM系统(开源方案)
# 导入现网IP地址数据
# 建立变更审批流程

# NetBox API批量导入
import requests

url = "https://netbox.example.com/api/ipam/ip-addresses/"
headers = {"Authorization": "Token YOUR_TOKEN"}

ip_data = [
    {"address": "10.100.1.10/24", "dns_name": "web-server-01", "status": "active"},
    {"address": "10.100.1.11/24", "dns_name": "web-server-02", "status": "active"},
    # ... 更多地址
]

for ip in ip_data:
    response = requests.post(url, json=ip, headers=headers)
    print(f"Imported {ip['address']}: {response.status_code}")

4.4 迁移后验证清单

IP地址迁移后必须按以下清单逐项验证,不可跳步:

即时检查(迁移后30分钟内):

检查项 验证方法 通过标准
新IP可达性 ping -c 5 新IP 0丢包,延迟<5ms
业务端口可用性 telnet 新IP 端口 端口可连接
DNS解析 nslookup 主机名 返回新IP
防火墙策略 display firewall session table 会话正常建立
监控数据 监控系统查看 新IP数据采集正常
日志系统 查看syslog 无异常报错

持续检查(迁移后1周内):

检查项 验证方法 通过标准
性能对比 监控平台对比迁移前后指标 延迟/吞吐无退化
间歇性问题 7×24小时监控 无偶发断连
旧IP流量 流量镜像分析 旧IP流量归零
用户反馈 工单系统 无相关投诉

4.5 迁移风险与应对

风险 概率 影响 应对措施
DNS缓存未刷新 高 中 TTL设为300秒,迁移前flush
防火墙ACL遗漏 中 高 IPAM自动生成ACL变更脚本
应用硬编码IP 低 高 代码扫描工具预先排查
数据库连接池失效 中 高 应用重启重连,灰度分批
存储多路径断裂 低 极高 保留旧IP多路径7天
配置回滚超时 低 极高 30分钟回滚SLA,预演回滚流程

4.7 Python实战:IP规划生成器

以下IPPlanner类自动化生成子网分配方案、检测地址重叠、生成最优汇总路由。

#!/usr/bin/env python3
"""
IP规划生成器
自动分配子网、检测重叠、生成汇总路由
"""
import ipaddress
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Set, Tuple


@dataclass
class SubnetAllocation:
    """子网分配"""
    subnet: ipaddress.IPv4Network
    site: str          # 站点名称
    purpose: str       # 用途描述
    vlan_id: int       # VLAN ID
    capacity: int = 0  # 总容量
    assigned: int = 0  # 已分配数量
    tags: List[str] = field(default_factory=list)

    @property
    def utilization(self) -> float:
        if self.capacity == 0:
            return 0.0
        return self.assigned / self.capacity * 100

    @property
    def usable_hosts(self) -> int:
        return max(0, self.subnet.num_addresses - 2)


@dataclass
class SummaryRoute:
    """汇总路由"""
    aggregate: ipaddress.IPv4Network
    components: List[ipaddress.IPv4Network]
    is_exact: bool     # 是否为精确汇总(无额外地址浪费)


class IPPlanner:
    """IP地址规划器"""

    def __init__(self):
        self.allocations: List[SubnetAllocation] = []
        self._all_networks: List[ipaddress.IPv4Network] = []

    def add_subnet(self,
                   cidr: str,
                   site: str,
                   purpose: str,
                   vlan_id: int,
                   assigned: int = 0,
                   tags: List[str] = None) -> SubnetAllocation:
        """
        添加子网分配
        :param cidr: CIDR格式子网,如10.1.1.0/24
        :param site: 站点名称
        :param purpose: 用途
        :param vlan_id: VLAN ID
        :param assigned: 已分配地址数
        :param tags: 标签列表
        :return: 分配对象
        """
        network = ipaddress.IPv4Network(cidr, strict=False)
        allocation = SubnetAllocation(
            subnet=network,
            site=site,
            purpose=purpose,
            vlan_id=vlan_id,
            capacity=network.num_addresses - 2,
            assigned=assigned,
            tags=tags or []
        )
        self.allocations.append(allocation)
        self._all_networks.append(network)
        return allocation

    def check_overlaps(self) -> List[Dict]:
        """
        检测子网重叠
        :return: 重叠信息列表
        """
        overlaps = []
        networks = [(a.subnet, a) for a in self.allocations]

        for i in range(len(networks)):
            for j in range(i + 1, len(networks)):
                net_i, alloc_i = networks[i]
                net_j, alloc_j = networks[j]

                if net_i.overlaps(net_j):
                    overlap_type = "完全包含" if net_i.supernet_of(net_j) or net_j.supernet_of(net_i) else "部分重叠"
                    overlaps.append({
                        "subnet1": str(net_i),
                        "site1": alloc_i.site,
                        "purpose1": alloc_i.purpose,
                        "subnet2": str(net_j),
                        "site2": alloc_j.site,
                        "purpose2": alloc_j.purpose,
                        "overlap_type": overlap_type
                    })

        return overlaps

    def generate_summary_routes(self,
                                group_by: str = "site"
                                ) -> List[SummaryRoute]:
        """
        生成汇总路由
        :param group_by: 分组依据(site/purpose)
        :return: 汇总路由列表
        """
        groups: Dict[str, List[ipaddress.IPv4Network]] = {}
        for alloc in self.allocations:
            key = getattr(alloc, group_by)
            if key not in groups:
                groups[key] = []
            groups[key].append(alloc.subnet)

        summaries = []
        for group_name, subnets in groups.items():
            if len(subnets) == 1:
                summaries.append(SummaryRoute(
                    aggregate=subnets[0],
                    components=[subnets[0]],
                    is_exact=True
                ))
            else:
                # 计算最小覆盖超网
                try:
                    aggregate = ipaddress.summarize_address_range(
                        min(s.network_address for s in subnets),
                        max(s.broadcast_address for s in subnets)
                    )
                    agg_list = list(aggregate)
                    # 检查是否为精确汇总
                    is_exact = (len(agg_list) == 1 and
                               all(s.subnet_of(agg_list[0]) for s in subnets) and
                               agg_list[0].num_addresses ==
                               sum(s.num_addresses for s in subnets))
                    summaries.append(SummaryRoute(
                        aggregate=agg_list[0],
                        components=subnets,
                        is_exact=is_exact
                    ))
                except ValueError:
                    # 无法汇总(非连续地址)
                    for s in subnets:
                        summaries.append(SummaryRoute(
                            aggregate=s,
                            components=[s],
                            is_exact=True
                        ))

        return summaries

    def generate_allocation_report(self) -> str:
        """生成分配报告"""
        lines = []
        lines.append("=" * 70)
        lines.append("IP地址分配报告")
        lines.append("=" * 70)
        lines.append(f"总子网数: {len(self.allocations)}")
        lines.append(f"总IP容量: {sum(a.capacity for a in self.allocations):,}")
        lines.append(f"已分配: {sum(a.assigned for a in self.allocations):,}")
        overall_util = (sum(a.assigned for a in self.allocations) /
                       max(1, sum(a.capacity for a in self.allocations)) * 100)
        lines.append(f"总体利用率: {overall_util:.1f}%")

        lines.append(f"\n{'站点':<15} {'子网':<20} {'VLAN':<6} {'用途':<15} "
                     f"{'容量':>6} {'已用':>6} {'利用率':>8}")
        lines.append("-" * 80)

        for a in sorted(self.allocations, key=lambda x: x.subnet.network_address):
            lines.append(
                f"{a.site:<15} {str(a.subnet):<20} {a.vlan_id:<6} "
                f"{a.purpose:<15} {a.capacity:>6} {a.assigned:>6} "
                f"{a.utilization:>7.1f}%"
            )

        # 重叠检查
        overlaps = self.check_overlaps()
        if overlaps:
            lines.append(f"\n⚠ 发现{len(overlaps)}处地址重叠!")
            for ov in overlaps:
                lines.append(
                    f"  {ov['subnet1']} ({ov['site1']}/{ov['purpose1']}) "
                    f"↔ {ov['subnet2']} ({ov['site2']}/{ov['purpose2']}) "
                    f"[{ov['overlap_type']}]"
                )
        else:
            lines.append("\n✓ 无地址重叠")

        # 汇总路由
        summaries = self.generate_summary_routes("site")
        lines.append(f"\n汇总路由(按站点):")
        for s in summaries:
            exact_mark = "✓精确" if s.is_exact else "✗有浪费"
            lines.append(f"  {s.aggregate} ({exact_mark}) "
                        f"<-- {len(s.components)}个子网")

        return "\n".join(lines)

    def export_to_csv(self, filename: str = "ip_allocation.csv"):
        """导出为CSV文件"""
        import csv
        with open(filename, "w", newline="", encoding="utf-8") as f:
            writer = csv.writer(f)
            writer.writerow(["Subnet", "Site", "Purpose", "VLAN_ID",
                            "Capacity", "Assigned", "Utilization", "Tags"])
            for a in self.allocations:
                writer.writerow([
                    str(a.subnet), a.site, a.purpose, a.vlan_id,
                    a.capacity, a.assigned, f"{a.utilization:.1f}%",
                    ";".join(a.tags)
                ])


# ========== 使用示例 ==========
if __name__ == "__main__":
    planner = IPPlanner()

    # 北京总部
    planner.add_subnet("10.1.1.0/24", "北京总部", "办公网-1F", 1001, 180, ["office", "floor1"])
    planner.add_subnet("10.1.2.0/24", "北京总部", "办公网-2F", 1002, 200, ["office", "floor2"])
    planner.add_subnet("10.1.3.0/24", "北京总部", "办公网-3F", 1003, 150, ["office", "floor3"])
    planner.add_subnet("10.1.10.0/24", "北京总部", "服务器", 2001, 30, ["server"])
    planner.add_subnet("10.1.20.0/24", "北京总部", "管理网", 3001, 15, ["mgmt"])

    # 上海分公司
    planner.add_subnet("10.2.1.0/24", "上海分公司", "办公网-1F", 1101, 120, ["office"])
    planner.add_subnet("10.2.2.0/24", "上海分公司", "办公网-2F", 1102, 80, ["office"])
    planner.add_subnet("10.2.10.0/24", "上海分公司", "服务器", 2101, 20, ["server"])

    # 数据中心
    planner.add_subnet("10.100.1.0/24", "北京DC", "Spine-Leaf互联", 4001, 50, ["dc", "underlay"])
    planner.add_subnet("10.100.2.0/23", "北京DC", "服务器接入", 4002, 400, ["dc", "server"])
    planner.add_subnet("10.100.10.0/24", "北京DC", "存储网络", 4010, 10, ["dc", "storage"])

    print(planner.generate_allocation_report())
    planner.export_to_csv()

4.8 Python实战:IP迁移辅助工具

以下MigrationPlanner类生成迁移脚本、检查清单和回滚预案。

#!/usr/bin/env python3
"""
IP迁移辅助工具
生成迁移脚本、检查清单和回滚预案
"""
import ipaddress
from dataclasses import dataclass, field
from typing import Dict, List, Optional
from datetime import datetime, timedelta


@dataclass
class MigrationItem:
    """单个迁移项"""
    device: str            # 设备名
    interface: str         # 接口名
    old_ip: str            # 旧IP(CIDR格式)
    new_ip: str            # 新IP(CIDR格式)
    migration_phase: int   # 迁移阶段(1=首批, 2=二批...)
    priority: str = "normal"  # critical/normal/low
    dependencies: List[str] = field(default_factory=list)  # 依赖的设备
    rollback_config: str = ""  # 回滚配置


@dataclass
class MigrationPlan:
    """迁移计划"""
    name: str
    description: str
    items: List[MigrationItem] = field(default_factory=list)
    start_time: Optional[datetime] = None
    rollback_deadline: Optional[datetime] = None


class MigrationPlanner:
    """IP迁移计划器"""

    def __init__(self, plan_name: str, description: str = ""):
        self.plan = MigrationPlan(
            name=plan_name,
            description=description,
            start_time=datetime.now(),
            rollback_deadline=datetime.now() + timedelta(hours=2)
        )
        self.checklist: List[Dict] = []

    def add_item(self,
                 device: str,
                 interface: str,
                 old_ip: str,
                 new_ip: str,
                 phase: int = 1,
                 priority: str = "normal",
                 dependencies: List[str] = None) -> MigrationItem:
        """添加迁移项"""
        old_net = ipaddress.IPv4Interface(old_ip)
        new_net = ipaddress.IPv4Interface(new_ip)

        item = MigrationItem(
            device=device,
            interface=interface,
            old_ip=str(old_net),
            new_ip=str(new_net),
            migration_phase=phase,
            priority=priority,
            dependencies=dependencies or [],
            rollback_config=self._gen_rollback_config(device, interface, old_ip)
        )
        self.plan.items.append(item)
        return item

    def _gen_rollback_config(self, device: str, interface: str, old_ip: str) -> str:
        """生成回滚配置"""
        return (
            f"system-view\n"
            f"interface {interface}\n"
            f"undo ip address\n"
            f"ip address {old_ip}\n"
            f"quit\n"
            f"save force"
        )

    def generate_migration_script(self, phase: int = None) -> str:
        """
        生成华为设备迁移脚本
        :param phase: 指定阶段,None=全部
        :return: 脚本内容
        """
        lines = []
        lines.append("#" + "=" * 60)
        lines.append(f"# IP迁移脚本 - {self.plan.name}")
        lines.append(f"# 生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        lines.append("#" + "=" * 60)

        items = self.plan.items
        if phase is not None:
            items = [i for i in items if i.migration_phase == phase]

        # 按优先级排序:critical > normal > low
        priority_order = {"critical": 0, "normal": 1, "low": 2}
        items = sorted(items, key=lambda x: (x.migration_phase, priority_order.get(x.priority, 1)))

        current_device = None
        for item in items:
            if item.device != current_device:
                if current_device is not None:
                    lines.append("quit  # 退出前一台设备\n")
                lines.append(f"\n# =====================")
                lines.append(f"# 设备: {item.device}")
                lines.append(f"# =====================")
                lines.append(f"ssh {item.device}")
                lines.append("system-view")
                current_device = item.device

            lines.append(f"\n# [{item.priority.upper()}] {item.interface}: "
                        f"{item.old_ip} -> {item.new_ip}")
            lines.append(f"interface {item.interface}")
            lines.append(f"# 先添加新IP(双栈并行模式)")
            lines.append(f"ip address {item.new_ip} sub")
            lines.append(f"quit")

        if current_device:
            lines.append("\nquit  # 退出SSH\n")

        lines.append("\n# 验证命令(每台设备执行)")
        lines.append("display ip interface brief")
        lines.append("display ip routing-table")

        return "\n".join(lines)

    def generate_checklist(self) -> str:
        """生成迁移检查清单"""
        lines = []
        lines.append("=" * 60)
        lines.append(f"迁移检查清单 - {self.plan.name}")
        lines.append("=" * 60)

        # 前置检查
        lines.append("\n## 迁移前检查(Pre-Check)")
        pre_checks = [
            {"item": "备份所有设备当前配置", "status": "[ ]"},
            {"item": "确认变更窗口已审批", "status": "[ ]"},
            {"item": "确认回滚预案已测试", "status": "[ ]"},
            {"item": "通知相关业务团队", "status": "[ ]"},
            {"item": "监控系统就绪(Telemetry/SNMP)", "status": "[ ]"},
            {"item": "确认所有目标设备SSH可达", "status": "[ ]"},
            {"item": "记录迁移前路由表快照", "status": "[ ]"},
            {"item": "记录迁移前BGP/OSPF邻居状态", "status": "[ ]"},
        ]
        for check in pre_checks:
            lines.append(f"  {check['status']} {check['item']}")

        # 迁移中检查(按阶段)
        phases = sorted(set(i.migration_phase for i in self.plan.items))
        for phase in phases:
            phase_items = [i for i in self.plan.items if i.migration_phase == phase]
            lines.append(f"\n## 阶段{phase}:{len(phase_items)}个迁移项")
            for item in phase_items:
                lines.append(f"  [ ] {item.device}/{item.interface}: "
                            f"{item.old_ip} -> {item.new_ip} "
                            f"[{item.priority}]")
                lines.append(f"      ├─ 添加新IP: done")
                lines.append(f"      ├─ 验证新IP可达: [ ]")
                lines.append(f"      ├─ 业务切换验证: [ ]")
                lines.append(f"      └─ 删除旧IP: [ ]")

        # 后置检查
        lines.append(f"\n## 迁移后检查(Post-Check)")
        post_checks = [
            {"item": "所有新IP可达(ping测试)", "status": "[ ]"},
            {"item": "路由表正确(无黑洞/次优)", "status": "[ ]"},
            {"item": "BGP/OSPF邻居状态正常", "status": "[ ]"},
            {"item": "业务流量切换完成", "status": "[ ]"},
            {"item": "监控系统数据采集正常", "status": "[ ]"},
            {"item": "旧IP流量归零", "status": "[ ]"},
            {"item": "更新IPAM系统(NetBox)", "status": "[ ]"},
            {"item": "更新防火墙ACL(如有)", "status": "[ ]"},
        ]
        for check in post_checks:
            lines.append(f"  {check['status']} {check['item']}")

        return "\n".join(lines)

    def generate_rollback_plan(self) -> str:
        """生成回滚预案"""
        lines = []
        lines.append("=" * 60)
        lines.append(f"回滚预案 - {self.plan.name}")
        lines.append(f"回滚截止时间: {self.plan.rollback_deadline}")
        lines.append("=" * 60)

        lines.append("\n## 回滚触发条件(满足任一即触发)")
        lines.append("  1. 迁移后30分钟内业务中断超过5%")
        lines.append("  2. 核心设备路由表异常(黑洞路由>10条)")
        lines.append("  3. BGP邻居大面积Down(>3个)")
        lines.append("  4. 无法在15分钟内定位问题根因")

        lines.append("\n## 回滚步骤")
        lines.append("  Step 1: 停止所有未完成的迁移操作")
        lines.append("  Step 2: 按优先级降序回滚(critical优先)")
        lines.append("  Step 3: 逐台设备执行回滚配置")

        # 按设备分组
        devices = {}
        for item in reversed(self.plan.items):  # 逆序回滚
            if item.device not in devices:
                devices[item.device] = []
            devices[item.device].append(item)

        for device, items in devices.items():
            lines.append(f"\n  ### 设备: {device}")
            for item in items:
                lines.append(f"    # 回滚 {item.interface}")
                lines.append(f"    {item.rollback_config}")

        lines.append("\n## 回滚验证")
        lines.append("  1. 旧IP全部恢复可达")
        lines.append("  2. 路由表与迁移前快照一致")
        lines.append("  3. 业务流量恢复正常")

        return "\n".join(lines)

    def generate_risk_assessment(self) -> str:
        """生成风险评估报告"""
        critical_count = sum(1 for i in self.plan.items if i.priority == "critical")
        total_items = len(self.plan.items)

        # 风险评估矩阵
        lines = []
        lines.append("=" * 60)
        lines.append("迁移风险评估报告")
        lines.append("=" * 60)
        lines.append(f"迁移项总数: {total_items}")
        lines.append(f"关键项(critical): {critical_count}")
        lines.append(f"普通项(normal): {total_items - critical_count - sum(1 for i in self.plan.items if i.priority == 'low')}")

        lines.append(f"\n{'风险项':<25} {'概率':<8} {'影响':<8} {'风险等级':<10}")
        lines.append("-" * 55)

        risks = [
            ("新IP配置错误导致中断", "低", "极高", "高"),
            ("迁移过程旧IP被删除过早", "中", "极高", "高"),
            ("防火墙ACL未同步更新", "中", "高", "中"),
            ("DNS缓存未刷新", "高", "中", "中"),
            ("BGP/OSPF邻居短暂中断", "中", "中", "中"),
            ("应用连接池不重连", "低", "高", "低"),
            ("回滚超时无法恢复", "极低", "极高", "低"),
        ]

        for risk, prob, impact, level in risks:
            lines.append(f"{risk:<25} {prob:<8} {impact:<8} {level:<10}")

        # 风险缓解措施
        lines.append("\n风险缓解措施:")
        lines.append("  1. 双栈并行:先添加新IP,验证后再删除旧IP")
        lines.append("  2. 每步验证:每个迁移项完成后立即验证可达性")
        lines.append("  3. 30分钟回滚SLA:超过30分钟未完成回滚即上报升级")
        lines.append("  4. 变更窗口:业务低谷期(凌晨00:00-04:00)执行")
        lines.append("  5. 专人值守:迁移期间网络、应用、DBA三方联合值守")

        return "\n".join(lines)


# ========== 使用示例 ==========
if __name__ == "__main__":
    planner = MigrationPlanner(
        plan_name="北京总部办公网IP迁移",
        description="将192.168.x.x迁移到10.1.x.x"
    )

    # Phase 1: 管理接口(关键)
    planner.add_item("Core-SW-01", "Vlanif100", "192.168.1.1/24",
                    "10.1.1.1/24", phase=1, priority="critical")
    planner.add_item("Core-SW-02", "Vlanif100", "192.168.1.2/24",
                    "10.1.1.2/24", phase=1, priority="critical")

    # Phase 2: 接入交换机
    planner.add_item("Access-SW-1F-01", "GigabitEthernet0/0/1",
                    "192.168.1.10/24", "10.1.1.10/24", phase=2, priority="normal")
    planner.add_item("Access-SW-1F-02", "GigabitEthernet0/0/1",
                    "192.168.1.11/24", "10.1.1.11/24", phase=2, priority="normal")
    planner.add_item("Access-SW-2F-01", "GigabitEthernet0/0/1",
                    "192.168.2.10/24", "10.1.2.10/24", phase=2, priority="normal",
                    dependencies=["Core-SW-01"])

    # Phase 3: 服务器
    planner.add_item("Server-Web-01", "eth0", "192.168.10.10/24",
                    "10.1.10.10/24", phase=3, priority="critical")
    planner.add_item("Server-DB-01", "eth0", "192.168.10.20/24",
                    "10.1.10.20/24", phase=3, priority="critical")

    print(planner.generate_migration_script(phase=1))
    print("\n" + planner.generate_checklist())
    print("\n" + planner.generate_rollback_plan())
    print("\n" + planner.generate_risk_assessment())

迁移风险评估表:

风险类别 风险项 概率 影响 风险等级 缓解措施 责任人
技术风险 新IP配置语法错误 低 极高 高 迁移前在模拟环境测试脚本 实施工程师
技术风险 双栈并行期间路由震荡 中 高 中 配置路由抑制时间,启用BFD 网络工程师
技术风险 设备不支持辅助IP 低 极高 低 提前确认设备型号和VRP版本 技术负责人
业务风险 数据库连接池不自动重连 中 高 中 迁移后重启应用服务 应用运维
业务风险 DNS解析缓存旧IP 高 中 中 迁移前TTL降至60s,迁移后flush 系统管理员
安全风险 防火墙ACL未同步更新 中 高 中 IPAM自动生成ACL变更脚本 安全工程师
流程风险 回滚超时(>30分钟) 低 极高 低 预演回滚流程,设定升级机制 项目经理
流程风险 变更窗口被占用 低 中 低 提前2周预约变更窗口 变更管理员
依赖风险 第三方设备不支持新IP 低 高 低 提前联系第三方确认 项目经理
依赖风险 监控工具未适配新IP段 中 中 中 迁移前更新监控配置 监控工程师

总体风险评估:中低风险。关键控制点:双栈并行策略确保任何时刻旧IP可用,30分钟回滚SLA兜底。

4.6 项目成果

指标 优化前 优化后 改善
核心路由条目 10,000+ 500 ↓95%
地址重叠 15处 0 消除
地址利用率 35% 72% ↑106%
地址文档 无 完整(NetBox) 新建
IPv6支持 无 双栈 新建
地址变更审批 无流程 标准化流程 新建

五、经验总结

序号 经验类别 经验内容 适用场景
1 审计先行 IP重规划必须先完成全面审计,不可盲目开始 项目管理
2 自动化采集 使用脚本自动采集全网IP,避免人工遗漏 自动化
3 层次化设计 地址分配必须按层次设计,确保可汇总 地址规划
4 预留空间 每个地址段预留50%扩展空间 容量规划
5 迁移策略 核心业务使用双栈并行,降低风险 迁移方案
6 IPAM系统 必须部署IPAM系统管理地址分配 地址管理
7 变更流程 IP地址变更必须有标准化审批流程 流程管理
8 IPv6并行 IP重规划时同步完成IPv6规划 演进规划
9 文档沉淀 地址分配文档必须实时更新 文档管理
10 定期审计 每季度进行一次IP地址审计 运维管理

附录:IP地址编码规则模板

一个标准的层次化IP编码规则,能够支撑数千台设备、数万端口的地址管理。

网络层次编码:

层次 位长 范围 用途 示例
站点(Region) 4bit 0-15 区分不同数据中心 10.0.x.x=北京, 10.16.x.x=上海
楼层(Floor) 4bit 0-15 区分机房楼层 10.0.0.x=1F, 10.0.16.x=2F
机架(Rack) 8bit 0-255 区分机架 10.0.0.0=Rack01
设备(Device) 4bit 0-15 区分设备 10.0.0.0=Spine, 10.0.0.1=Leaf-01
端口(Port) 12bit 0-4095 区分端口 10.0.0.1.0=Leaf-01/Port1

功能子网划分:

用途 子网前缀 掩码 地址范围 规划说明
管理网 10.255.0.0/16 /24 per DC 10.255.0.0-10.255.255.255 带外管理,独立VLAN
Loopback 10.254.0.0/16 /32 per device 10.254.0.1-10.254.255.254 设备Loopback接口
互联链路 10.0.0.0/8 /31 per link Point-to-Point 节省地址
服务器接入 172.16.0.0/12 /24 per rack 每机架一个子网 易于汇总
VPN业务 100.64.0.0/10 /24 per VRF 共享地址空间 CGNAT地址

IPAM系统字段规范:

字段 必填 说明 示例
subnet ✓ CIDR格式子网 172.16.1.0/24
vlan_id ✓ 对应VLAN 1001
site ✓ 所属站点 BJ-DC1
floor ✓ 所属楼层 3F
rack ✓ 所属机架 Rack-042
switch ✓ 接入交换机 Leaf-3F-042
purpose ✓ 用途 Server-Access
status ✓ 状态 Active/Reserved/Decommissioned
assigned_percentage - 已分配比例 65%
owner ✓ 负责人 network-team

思考

  1. 在进行IP地址重规划时,如果某些老旧系统不支持CIDR(只支持点分十进制掩码),如何处理?
  2. IPv4地址耗尽是大势所趋,企业在IPv4向IPv6迁移过程中,双栈、NAT64/DNS64、翻译等方案如何选择?
  3. IPAM系统应该如何选择?开源方案(NetBox、phpIPAM)与商业方案(Infoblox、SolarWinds)各有什么优缺点?

下篇预告:第349篇:老旧网络设备替换与架构升级方案