第315篇:IPAM 地址管理

关键词

IPAM、IP 地址管理、子网规划、DHCP 集成、DNS 集成、IP 资源管理、网络自动化、IP 生命周期


一、IPAM 概述

1.1 什么是 IPAM

IPAM(IP Address Management)是 IP 地址的 全生命周期管理,涵盖规划、分配、跟踪和回收:

没有 IPAM 的痛点:

  ┌──────────────────────────────────────────┐
  │  Excel 管理 IP 的常见问题:                │
  │  ┌─ 多人编辑 → 数据不一致                │
  │  ├─ 手动更新 → 不及时                    │
  │  ├─ 无冲突检测 → IP 重复分配            │
  │  ├─ 无历史记录 → 无法追溯               │
  │  ├─ 扩展困难 → 百级设备还行,万级崩溃    │
  │  └─ 无自动化 → 申请/分配/回收全手工      │
  │                                            │
  │  IP 管理失控的后果:                       │
  │  ┌─ IP 冲突 → 业务中断                   │
  │  ├─ 子网碎片 → 地址利用率低              │
  │  ├─ 无法定位 → 故障排查困难              │
  │  └─ 僵尸 IP → 安全隐患                   │
  └──────────────────────────────────────────┘

  IPAM 的价值:
  ┌─ 统一管理:所有 IP 资源集中管理
  ├─ 自动分配:减少人工操作
  ├─ 冲突检测:避免 IP 重复
  ├─ 利用率分析:优化 IP 空间
  ├─ 历史追溯:谁在何时使用了哪个 IP
  └─ API 集成:与自动化工具联动

1.2 IPAM 核心功能

IPAM 的核心能力:

  1. 子网管理
  ┌─ 规划子网层次(VLSM)
  ├─ 子网划分与聚合
  ├─ 可用 IP 自动计算
  └─ 利用率统计

  2. IP 分配
  ┌─ 静态分配(服务器/网络设备)
  ├─ 动态分配(DHCP 范围)
  ├─ 保留地址(网关/VIP)
  └─ 分配审批工作流

  3. DNS 集成
  ┌─ 自动创建/删除 A/PTR 记录
  ├─ 反向查找区域管理
  └─ DNS 视图管理

  4. DHCP 集成
  ┌─ 作用域配置同步
  ├─ 预留与排除范围
  └─ 租约信息同步

  5. 审计与报告
  ┌─ 变更历史
  ├─ 利用率报告
  ├─ 到期提醒
  └─ 合规审计

二、IPAM 数据模型

2.1 层次结构

IPAM 数据层次:

组织/客户(Organization/Tenant) └─ 站点(Site) └─ VRF/网络视图(VRF) └─ 子网(Subnet) ├─ 可用 IP 范围(Range) ├─ 已分配 IP(IP Address) │ ├─ 设备(Device) │ ├─ 接口(Interface) │ └─ DNS 记录(DNS Record) └─ DHCP 范围(DHCP Pool)

示例: | Tenant: Corp ├─ Site: Beijing-DC └─ Site: Shanghai-DC | ├─ VRF: PROD └─ VRF: DEV | ├─ 10.0.0.0/8 └─ 172.16.0.0/12 | ├─ 10.0.1.0/24 (MGMT) └─ 10.0.2.0/24 (OFFICE) | ├─ 10.0.1.1 (核心网关) ├─ 10.0.1.2 (CORE-SW01) └─ 10.0.1.10-20 (服务器) | | --- | --- | --- | --- | --- |

2.2 数据结构设计

#!/usr/bin/env python3
# ipam_model.py — IPAM 数据模型

from datetime import datetime
from ipaddress import IPv4Network, IPv4Address
from typing import Optional, List
import json

class IPSubnet:
    """子网模型"""

    def __init__(
        self,
        network: str,
        name: str = "",
        vrf: str = "global",
        site: str = "",
        description: str = "",
    ):
        self.network = IPv4Network(network, strict=False)
        self.name = name or str(self.network)
        self.vrf = vrf
        self.site = site
        self.description = description
        self.allocations: List[IPAllocation] = []
        self.created_at = datetime.now()
        self.updated_at = datetime.now()

    @property
    def total_ips(self) -> int:
        """总 IP 数"""
        return self.network.num_addresses

    @property
    def used_ips(self) -> int:
        """已用 IP 数"""
        return len(self.allocations)

    @property
    def utilization(self) -> float:
        """利用率"""
        if self.total_ips == 0:
            return 0.0
        return round(self.used_ips / self.total_ips * 100, 1)

    @property
    def available_ips(self) -> int:
        """可用 IP 数"""
        used = set(a.ip for a in self.allocations)
        # 减去网络地址和广播地址
        reserved = {
            str(self.network.network_address),
            str(self.network.broadcast_address),
        }
        total_available = self.total_ips - len(used | reserved)
        return total_available

    def allocate(self, ip: str, hostname: str, owner: str = "") -> bool:
        """分配 IP"""
        try:
            addr = IPv4Address(ip)
            if addr not in self.network:
                return False

            # 检查是否已分配
            for a in self.allocations:
                if a.ip == ip:
                    return False

            alloc = IPAllocation(
                ip=ip,
                hostname=hostname,
                owner=owner,
            )
            self.allocations.append(alloc)
            self.updated_at = datetime.now()
            return True
        except (ValueError, TypeError):
            return False

    def release(self, ip: str) -> bool:
        """释放 IP"""
        for i, a in enumerate(self.allocations):
            if a.ip == ip:
                self.allocations.pop(i)
                self.updated_at = datetime.now()
                return True
        return False

    def find_ip(self, hostname: str) -> Optional[str]:
        """通过主机名查找 IP"""
        for a in self.allocations:
            if a.hostname == hostname:
                return a.ip
        return None

    def to_dict(self) -> dict:
        """转为字典"""
        return {
            "network": str(self.network),
            "name": self.name,
            "vrf": self.vrf,
            "site": self.site,
            "description": self.description,
            "total_ips": self.total_ips,
            "used_ips": self.used_ips,
            "available_ips": self.available_ips,
            "utilization": self.utilization,
            "allocations": [a.to_dict() for a in self.allocations],
        }


class IPAllocation:
    """IP 分配记录"""

    def __init__(
        self,
        ip: str,
        hostname: str,
        owner: str = "",
        description: str = "",
    ):
        self.ip = ip
        self.hostname = hostname
        self.owner = owner
        self.description = description
        self.allocated_at = datetime.now()
        self.ttl = 86400  # 默认 1 天

    def to_dict(self) -> dict:
        return {
            "ip": self.ip,
            "hostname": self.hostname,
            "owner": self.owner,
            "description": self.description,
            "allocated_at": self.allocated_at.isoformat(),
        }


class IPAM:
    """IP 地址管理器"""

    def __init__(self):
        self.vrfs: dict = {}       # VRF → {site → [subnets]}
        self.subnets: list = []    # 所有子网

    def add_subnet(self, subnet: IPSubnet):
        """添加子网"""
        self.subnets.append(subnet)

        if subnet.vrf not in self.vrfs:
            self.vrfs[subnet.vrf] = {}
        if subnet.site not in self.vrfs[subnet.vrf]:
            self.vrfs[subnet.vrf][subnet.site] = []
        self.vrfs[subnet.vrf][subnet.site].append(subnet)

    def find_subnet(self, ip: str) -> Optional[IPSubnet]:
        """查找 IP 所属的子网"""
        try:
            addr = IPv4Address(ip)
            for subnet in self.subnets:
                if addr in subnet.network:
                    return subnet
        except ValueError:
            pass
        return None

    def get_utilization_report(self) -> dict:
        """利用率报告"""
        total_ips = sum(s.total_ips for s in self.subnets)
        used_ips = sum(s.used_ips for s in self.subnets)
        return {
            "total_subnets": len(self.subnets),
            "total_ips": total_ips,
            "used_ips": used_ips,
            "overall_utilization": round(used_ips / total_ips * 100, 1) if total_ips else 0,
            "by_vrf": {
                vrf: {
                    "subnets": len(sites),
                    "total": sum(
                        s.total_ips for site_subnets in sites.values()
                        for s in site_subnets
                    ),
                    "used": sum(
                        s.used_ips for site_subnets in sites.values()
                        for s in site_subnets
                    ),
                }
                for vrf, sites in self.vrfs.items()
            },
        }

    def to_json(self, path: str):
        """导出为 JSON"""
        data = {
            "subnets": [s.to_dict() for s in self.subnets],
            "report": self.get_utilization_report(),
        }
        with open(path, "w", encoding="utf-8") as f:
            json.dump(data, f, indent=2, ensure_ascii=False)

    @classmethod
    def from_json(cls, path: str) -> "IPAM":
        """从 JSON 导入"""
        ipam = cls()
        with open(path, "r", encoding="utf-8") as f:
            data = json.load(f)
        for s in data.get("subnets", []):
            subnet = IPSubnet(
                network=s["network"],
                name=s.get("name", ""),
                vrf=s.get("vrf", "global"),
                site=s.get("site", ""),
                description=s.get("description", ""),
            )
            for a in s.get("allocations", []):
                alloc = IPAllocation(
                    ip=a["ip"],
                    hostname=a["hostname"],
                    owner=a.get("owner", ""),
                    description=a.get("description", ""),
                )
                subnet.allocations.append(alloc)
            ipam.add_subnet(subnet)
        return ipam


# ===== 测试使用 =====
if __name__ == "__main__":
    ipam = IPAM()

    # 添加子网
    mgmt = IPSubnet("10.0.1.0/24", "Management", site="Beijing-DC")
    mgmt.allocate("10.0.1.1", "CORE-SW01", "network-team")
    mgmt.allocate("10.0.1.2", "CORE-RT01", "network-team")
    mgmt.allocate("10.0.1.10", "monitor-server", "ops-team")
    ipam.add_subnet(mgmt)

    office = IPSubnet("10.0.2.0/24", "Office", site="Beijing-DC")
    office.allocate("10.0.2.1", "CORE-SW01-VLAN20", "network-team")
    ipam.add_subnet(office)

    # 报告
    report = ipam.get_utilization_report()
    print("=== IP 利用率报告 ===")
    print(f"总子网: {report['total_subnets']}")
    print(f"总 IP: {report['total_ips']}")
    print(f"已用 IP: {report['used_ips']}")
    print(f"整体利用率: {report['overall_utilization']}%")

    print("\n=== 子网详情 ===")
    for subnet in ipam.subnets:
        print(f"\n{subnet.name} ({subnet.network}):")
        print(f"  总 IP: {subnet.total_ips}, 已用: {subnet.used_ips}, "
              f"可用: {subnet.available_ips}, 利用率: {subnet.utilization}%")
        for alloc in subnet.allocations:
            print(f"  - {alloc.ip} → {alloc.hostname} ({alloc.owner})")

    # 查找 IP
    found = ipam.find_subnet("10.0.1.10")
    print(f"\n查找 10.0.1.10: {found.name if found else '未找到'}")

    # 导出
    ipam.to_json("ipam_data.json")
    print("\n✓ IPAM 数据已导出到 ipam_data.json")

三、IPAM 工具选型

3.1 主流 IPAM 工具

工具 类型 开源 API DHCP/DNS 适用规模
NetBox phpIPAM Infoblox SolarWinds BlueCat GestióIP DCIM IPAM 商业 商业 商业 IPAM 是 是 否 否 否 是 完善 有 完善 有 完善 有 模块化 集成 原生 集成 原生 无 中大型 中小型 大型企业 中大型 大型企业 中小型

NetBox(推荐首选): ┌─ 开源、社区活跃 ├─ DCIM + IPAM + 设备管理 ├─ REST API 完善 ├─ 与 Ansible/NAPALM 集成 └─ 适合网络自动化

3.2 NetBox 集成

#!/usr/bin/env python3
# netbox_ipam_sync.py — NetBox IP 管理集成

import requests
import json
from urllib.parse import urljoin

class NetBoxClient:
    """NetBox API 客户端"""

    def __init__(self, base_url: str, token: str):
        self.base_url = base_url.rstrip("/")
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Token {token}",
            "Accept": "application/json",
            "Content-Type": "application/json",
        })

    def _get(self, path: str, params: dict = None) -> dict:
        url = urljoin(self.base_url + "/", path.lstrip("/"))
        resp = self.session.get(url, params=params)
        resp.raise_for_status()
        return resp.json()

    def _post(self, path: str, data: dict) -> dict:
        url = urljoin(self.base_url + "/", path.lstrip("/"))
        resp = self.session.post(url, json=data)
        resp.raise_for_status()
        return resp.json()

    # ---- IPAM 操作 ----

    def list_prefixes(self, **filters) -> list:
        """获取前缀(子网)列表"""
        result = self._get("/api/ipam/prefixes/", params=filters)
        return result.get("results", [])

    def create_prefix(self, prefix: str, site: str = None,
                      vrf: str = None, description: str = ""):
        """创建前缀"""
        data = {
            "prefix": prefix,
            "description": description,
        }
        if site:
            data["site"] = self._get_site_id(site)
        if vrf:
            data["vrf"] = self._get_vrf_id(vrf)
        return self._post("/api/ipam/prefixes/", data)

    def list_ip_addresses(self, **filters) -> list:
        """获取 IP 地址列表"""
        result = self._get("/api/ipam/ip-addresses/", params=filters)
        return result.get("results", [])

    def create_ip_address(self, address: str, device: str = None,
                          interface: str = None,
                          description: str = ""):
        """创建 IP 地址"""
        data = {
            "address": address,
            "description": description,
        }
        if device and interface:
            data["assigned_object_type"] = "dcim.interface"
            data["assigned_object_id"] = self._get_interface_id(
                device, interface
            )
        return self._post("/api/ipam/ip-addresses/", data)

    def _get_site_id(self, name: str) -> int:
        sites = self._get("/api/dcim/sites/", {"name": name})
        return sites["results"][0]["id"]

    def _get_vrf_id(self, name: str) -> int:
        vrfs = self._get("/api/ipam/vrfs/", {"name": name})
        return vrfs["results"][0]["id"]

    def _get_interface_id(self, device: str, interface: str) -> int:
        devices = self._get("/api/dcim/devices/", {"name": device})
        device_id = devices["results"][0]["id"]
        ifaces = self._get(
            "/api/dcim/interfaces/",
            {"device_id": device_id, "name": interface},
        )
        return ifaces["results"][0]["id"]


# 使用示例
def sync_device_to_netbox(device_info: dict, nb_client: NetBoxClient):
    """同步设备 IP 到 NetBox"""
    hostname = device_info["hostname"]
    interfaces = device_info.get("interfaces", [])

    for intf in interfaces:
        if "ip" not in intf:
            continue

        ip_addr = f"{intf['ip']}/{intf.get('prefix', 24)}"
        try:
            result = nb_client.create_ip_address(
                address=ip_addr,
                device=hostname,
                interface=intf["name"],
                description=intf.get("description", ""),
            )
            print(f"✓ {hostname} {intf['name']}: {ip_addr} 已同步")
        except Exception as e:
            print(f"✗ {hostname} {intf['name']}: 同步失败 {e}")


if __name__ == "__main__":
    # 配置
    NB_URL = "http://netbox.corp.com"
    NB_TOKEN = "your-api-token-here"

    client = NetBoxClient(NB_URL, NB_TOKEN)

    # 查询前缀
    prefixes = client.list_prefixes()
    print("=== NetBox 前缀列表 ===")
    for p in prefixes[:5]:
        print(f"  {p['prefix']} - {p.get('description', '')} "
              f"[{p['status']['label']}]")

    # 创建设备 IP
    device_intf = {
        "hostname": "ACC-SW01",
        "interfaces": [
            {"name": "GigabitEthernet0/0/1", "ip": "10.0.1.1", "prefix": 24},
        ],
    }
    sync_device_to_netbox(device_intf, client)

四、IPAM 自动化集成

4.1 自动发现与同步

#!/usr/bin/env python3
# ipam_auto_discover.py — IP 自动发现与同步

"""
从网络设备自动发现接口 IP,同步到 IPAM 系统
"""

import sys
import os
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))

from lib.connector import DeviceConnector
from lib.logger import setup_logger
import json
from concurrent.futures import ThreadPoolExecutor, as_completed
from ipaddress import IPv4Interface

log = setup_logger("ipam_discover")

def discover_device_ips(device: dict) -> dict:
    """发现设备接口 IP"""
    connector = DeviceConnector()
    conn = connector.connect_netmiko(device)
    if not conn:
        return device["name"], {"status": "failed"}

    interfaces = []
    try:
        # 获取 IP 接口信息
        output = conn.send_command("display ip interface brief")
        for line in output.split("\n"):
            parts = line.split()
            if len(parts) >= 4 and parts[0].startswith(("Gigabit", "Loop", "Eth", "Vlanif", "MEth")):
                iface_name = parts[0]
                ip_addr = parts[1]
                status = parts[3] if len(parts) > 3 else "down"

                # 过滤无效 IP
                if ip_addr and ip_addr != "unassigned" and ip_addr != "-":
                    interfaces.append({
                        "name": iface_name,
                        "ip": ip_addr,
                        "status": status,
                    })

        # 获取接口描述
        desc_output = conn.send_command("display interface description")
        desc_map = {}
        for line in desc_output.split("\n"):
            parts = line.split()
            if len(parts) >= 2:
                desc_map[parts[0]] = " ".join(parts[1:]) if len(parts) > 1 else ""

        for intf in interfaces:
            intf["description"] = desc_map.get(intf["name"], "")

        log.info(f"✓ {device['name']}: 发现 {len(interfaces)} 个 IP 接口")

    except Exception as e:
        log.error(f"✗ {device['name']}: {e}")
        return device["name"], {"status": "failed", "error": str(e)}
    finally:
        connector.close(conn)

    return device["name"], {
        "hostname": device["name"],
        "status": "success",
        "interfaces": interfaces,
    }

def generate_ipam_import(all_results: dict):
    """生成 IPAM 导入数据"""
    import_data = {
        "subnets": {},
        "allocations": [],
    }

    for name, result in all_results.items():
        if result.get("status") != "success":
            continue

        for intf in result.get("interfaces", []):
            ip_parts = intf["ip"].split("/")
            ip = ip_parts[0]
            prefix = ip_parts[1] if len(ip_parts) > 1 else "24"

            # 按 /24 子网分组(简化)
            subnet_key = f"{'.'.join(ip.split('.')[:3])}.0/{prefix}"
            if subnet_key not in import_data["subnets"]:
                # 推断子网,取前 3 段
                network = f"{'.'.join(ip.split('.')[:3])}.0/{prefix}"
                import_data["subnets"][subnet_key] = {
                    "network": network,
                    "name": f"Auto-Discovered-{subnet_key}",
                }

            import_data["allocations"].append({
                "device": result["hostname"],
                "interface": intf["name"],
                "ip": ip,
                "prefix": prefix,
                "description": intf.get("description", ""),
            })

    return import_data

def main():
    """主函数"""
    log.info("=== IP 自动发现开始 ===")
    connector = DeviceConnector()
    all_results = {}

    with ThreadPoolExecutor(max_workers=5) as ex:
        futures = {ex.submit(discover_device_ips, dev): dev
                   for dev in connector.devices}
        for future in as_completed(futures):
            name, result = future.result()
            all_results[name] = result

    # 生成导入数据
    import_data = generate_ipam_import(all_results)

    os.makedirs("output/ipam", exist_ok=True)
    with open("output/ipam/discovered_ips.json", "w") as f:
        json.dump(import_data, f, indent=2, ensure_ascii=False)

    log.info(f"发现完成: {len(all_results)} 台设备, "
             f"{len(import_data['allocations'])} 个 IP 地址")
    log.info(f"导入文件: output/ipam/discovered_ips.json")

if __name__ == "__main__":
    main()

4.2 DHCP 预留同步

#!/usr/bin/env python3
# dhcp_reservation_sync.py — DHCP 预留同步

"""
从 IPAM 获取静态分配列表,生成 DHCP 预留配置
"""

def generate_dhcp_reservations(ipam_allocations: list) -> str:
    """生成 DHCP 预留配置"""
    config_lines = [
        "# DHCP 服务器预留配置",
        "# 自动生成,请勿手动修改",
        f"# 生成时间: {__import__('datetime').datetime.now()}",
        "",
    ]

    for alloc in sorted(ipam_allocations, key=lambda x: x["ip"]):
        config_lines.extend([
            f"# {alloc.get('description', '')}",
            f"# 设备: {alloc.get('device', 'unknown')} / {alloc.get('interface', '')}",
            f"host {alloc['hostname']} {{",
            f"  hardware ethernet {alloc.get('mac', '00:00:00:00:00:00')};",
            f"  fixed-address {alloc['ip']};",
            f"  option host-name \"{alloc['hostname']}\";",
            f"}}",
            "",
        ])

    return "\n".join(config_lines)


def generate_dhcp_subnet(subnet: dict) -> str:
    """生成 DHCP 子网配置"""
    lines = [
        f"subnet {subnet['network']} netmask {subnet['netmask']} {{",
        f"  option routers {subnet['gateway']};",
        f"  option subnet-mask {subnet['netmask']};",
        f"  option domain-name-servers {subnet.get('dns', '8.8.8.8')};",
        f"  option domain-name \"{subnet.get('domain', 'corp.local')}\";",
        f"  range {subnet['range_start']} {subnet['range_end']};",
    ]

    if subnet.get("reservations"):
        lines.append("")
        for r in subnet["reservations"]:
            lines.append(f"  host {r['hostname']} {{")
            lines.append(f"    hardware ethernet {r['mac']};")
            lines.append(f"    fixed-address {r['ip']};")
            lines.append(f"  }}")

    lines.append("}")
    return "\n".join(lines)


# 示例
subnet_config = {
    "network": "10.0.20.0",
    "netmask": "255.255.255.0",
    "gateway": "10.0.20.1",
    "dns": "10.0.0.10",
    "domain": "office.corp.local",
    "range_start": "10.0.20.100",
    "range_end": "10.0.20.200",
    "reservations": [
        {"hostname": "printer-01", "mac": "aa:bb:cc:dd:ee:01", "ip": "10.0.20.10"},
        {"hostname": "printer-02", "mac": "aa:bb:cc:dd:ee:02", "ip": "10.0.20.11"},
        {"hostname": "camera-01", "mac": "aa:bb:cc:dd:ee:03", "ip": "10.0.20.20"},
    ],
}

print(generate_dhcp_subnet(subnet_config))

五、IP 生命周期管理

5.1 IP 申请与审批流程

IP 自动化管理流程:

  1. IP 申请
  ┌─ 工程师提交申请(IPAM Web/API)
  ├─ 填写:主机名、用途、子网、所属人
  └─ 系统自动检测是否可用

  2. 自动分配
  ┌─ IPAM 从指定子网分配可用 IP
  ├─ 创建 DNS A/PTR 记录
  ├─ 创建 DHCP 预留(如果需要)
  └─ 更新 CMDB/NetBox

  3. 配置下发
  ┌─ 自动生成设备配置
  ├─ 通过 Netmiko/Ansible 下发
  └─ 验证 IP 连通性

  4. 回收
  ┌─ 设备下线触发 IP 回收
  ├─ 清除 DNS 记录
  ├─ 清除 DHCP 预留
  └─ IP 回到可用池

  5. 审计
  ┌─ 定期扫描网络中实际使用的 IP
  ├─ 与 IPAM 数据对比
  ├─ 发现未记录 IP(僵尸 IP)
  └─ 生成利用率报告

5.2 IP 利用率分析

#!/usr/bin/env python3
# ipam_analysis.py — IP 利用率分析报告

from ipaddress import IPv4Network
from datetime import datetime

def analyze_subnet_usage(subnets: list) -> dict:
    """分析子网使用情况"""
    analysis = {
        "total_subnets": len(subnets),
        "total_ip_space": 0,
        "total_used": 0,
        "subnets_by_utilization": {
            "high (>80%)": [],
            "medium (50-80%)": [],
            "low (<50%)": [],
            "full (100%)": [],
        },
        "wasted_space": 0,
        "fragmentation": [],
    }

    for subnet in subnets:
        network = IPv4Network(subnet["network"])
        total = network.num_addresses
        used = subnet.get("used", 0)
        utilization = (used / total * 100) if total > 0 else 0

        analysis["total_ip_space"] += total
        analysis["total_used"] += used

        # 分类
        if utilization >= 100:
            analysis["subnets_by_utilization"]["full (100%)"].append(subnet)
        elif utilization > 80:
            analysis["subnets_by_utilization"]["high (>80%)"].append(subnet)
        elif utilization >= 50:
            analysis["subnets_by_utilization"]["medium (50-80%)"].append(subnet)
        else:
            analysis["subnets_by_utilization"]["low (<50%)"].append(subnet)

        # 碎片分析
        if used > 0 and utilization < 30 and total > 64:
            analysis["fragmentation"].append({
                "subnet": subnet["network"],
                "utilization": round(utilization, 1),
                "used": used,
                "available": total - used,
            })

    return analysis

# 示例
subnets_example = [
    {"network": "10.0.1.0/24", "used": 240},    # 93.8%
    {"network": "10.0.2.0/24", "used": 254},    # 100%
    {"network": "10.0.3.0/24", "used": 5},      # 2% — 碎片
    {"network": "10.0.10.0/24", "used": 128},   # 50%
    {"network": "10.0.20.0/24", "used": 45},    # 17.6% — 碎片
]

analysis = analyze_subnet_usage(subnets_example)
print("=== IP 利用率分析报告 ===")
print(f"总子网: {analysis['total_subnets']}")
print(f"总 IP 空间: {analysis['total_ip_space']}")
print(f"已使用: {analysis['total_used']}")
print(f"整体利用率: {analysis['total_used']/analysis['total_ip_space']*100:.1f}%")
print()

for category, subs in analysis["subnets_by_utilization"].items():
    print(f"{category}: {len(subs)} 个子网")
    for s in subs:
        print(f"  {s['network']} (已用 {s.get('used', 0)})")

if analysis["fragmentation"]:
    print(f"\n⚠ 碎片子网(利用率 < 30%):")
    for f in analysis["fragmentation"]:
        print(f"  {f['subnet']}: {f['utilization']}% 利用率, "
              f"{f['available']} 个可用 IP")

六、最佳实践

6.1 IPAM 实施规范

IPAM 实施最佳实践:

  1. 规划先行
  ┌─ 按业务/功能划分子网
  ├─ 预留扩展空间(利用率 < 70%)
  ├─ 统一 CIDR 规划
  └─ 文档化分配策略

  2. 数据质量
  ┌─ 每个 IP 必须有 Owner
  ├─ 记录用途/描述
  ├─ 定期扫描与实际对比
  └─ 清理僵尸 IP

  3. 自动化
  ┌─ 设备上线自动分配 IP
  ├─ 设备下线自动回收
  ├─ DNS/DHCP 自动同步
  └─ 与 CMDB/NetBox 集成

  4. 安全
  ┌─ IP 分配需要审批
  ├─ 敏感子网(DMZ/管理网)严格管控
  ├─ 审计日志记录所有操作
  └─ 定期安全扫描

6.2 常见问题

IPAM 常见问题:

  IP 冲突:
  ┌─ 原因:手工分配/未同步
  ├─ 解决:IPAM + 扫描 + 自动检测
  └─ 预防:统一通过 IPAM 分配

  子网耗尽:
  ┌─ 原因:未规划扩展空间
  ├─ 解决:VLSM 重新划分
  └─ 预防:利用率 > 70% 时预警

  DNS 不一致:
  ┌─ 原因:IPAM 与 DNS 不同步
  ├─ 解决:自动同步
  └─ 预防:IPAM 作为单一数据源

七、总结

IPAM 的核心价值:

  从 Excel 到平台化管理
  ┌─ 手工 → 自动
  ├─ 静态 → 实时
  ├─ 局部 → 全局
  └─ 被动 → 主动

  IPAM 自动化链路:
  IPAM (NetBox) → 自动分配 → DNS 同步 → DHCP 预留 → 配置下发
                    ↑           ↑          ↑           ↑
                  可用 IP     DNS 记录   MAC + IP   设备配置

  推荐工具链:
  IP 管理:NetBox(开源首选)
  自动化:Ansible + Python
  监控:Prometheus + Grafana
  集成:REST API

下篇预告:第316篇 — CI/CD 网络变更流水线,将介绍如何将 CI/CD 理念引入网络变更管理,实现配置的持续集成、自动测试和灰度发布。