第320篇：IP 地址管理（IPAM）自动化

关键词：IPAM、IP地址管理、子网划分、地址分配、冲突检测、NetBox、自动化分配、地址池管理

一、引言

第319篇介绍了配置备份与合规检查。在网络运维中，IP 地址管理（IP Address Management, IPAM）是又一个高频且易出错的操作：随着网络规模扩大，手工维护的 Excel 表格不仅效率低下，而且极易出现地址冲突、记录不一致、子网耗尽无人知晓等问题。

IPAM 自动化的核心目标是通过程序化管理 IP 地址的分配、回收、查询和冲突检测，确保地址使用的准确性和可追溯性。

二、核心概念

2.1 IPAM 核心功能

2.2 子网划分策略

2.3 IP 地址池类型

三、技术细节

3.1 IPAM 数据库设计

# scripts/ipam_core.py
import sqlite3
import ipaddress
from datetime import datetime
from typing import Optional, List, Dict, Tuple

class IPAMDatabase:
    """IPAM 核心数据库"""

    def __init__(self, db_path="ipam.db"):
        self.db_path = db_path
        self._init_db()

    def _init_db(self):
        """初始化数据库表"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()

        # 站点表
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS sites (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                name TEXT NOT NULL UNIQUE,
                region TEXT,
                description TEXT,
                created_at TEXT DEFAULT (datetime('now'))
            )
        ''')

        # 子网表（支持嵌套子网）
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS subnets (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                network TEXT NOT NULL,
                prefix_length INTEGER NOT NULL,
                parent_subnet_id INTEGER,
                site_id INTEGER,
                vlan_id INTEGER,
                purpose TEXT,
                description TEXT,
                status TEXT DEFAULT 'active',
                created_at TEXT DEFAULT (datetime('now')),
                UNIQUE(network, prefix_length),
                FOREIGN KEY (parent_subnet_id) REFERENCES subnets(id),
                FOREIGN KEY (site_id) REFERENCES sites(id)
            )
        ''')

        # IP 地址分配表
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS ip_allocations (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                ip_address TEXT NOT NULL,
                subnet_id INTEGER NOT NULL,
                mac_address TEXT,
                hostname TEXT,
                device_type TEXT,
                assigned_to TEXT,
                purpose TEXT,
                allocation_type TEXT DEFAULT 'static',
                lease_start TEXT,
                lease_end TEXT,
                status TEXT DEFAULT 'active',
                created_at TEXT DEFAULT (datetime('now')),
                updated_at TEXT DEFAULT (datetime('now')),
                FOREIGN KEY (subnet_id) REFERENCES subnets(id),
                UNIQUE(ip_address, status)
            )
        ''')

        # 创建索引
        cursor.execute('CREATE INDEX IF NOT EXISTS idx_ip ON ip_allocations(ip_address)')
        cursor.execute('CREATE INDEX IF NOT EXISTS idx_subnet ON ip_allocations(subnet_id)')
        cursor.execute('CREATE INDEX IF NOT EXISTS idx_status ON ip_allocations(status)')
        cursor.execute('CREATE INDEX IF NOT EXISTS idx_network ON subnets(network)')

        conn.commit()
        conn.close()

    # ========== 站点管理 ==========

    def add_site(self, name: str, region: str = None, description: str = None) -> int:
        """添加站点"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.execute(
            'INSERT OR IGNORE INTO sites (name, region, description) VALUES (?, ?, ?)',
            (name, region, description)
        )
        conn.commit()
        # 获取站点 ID
        cursor = conn.execute('SELECT id FROM sites WHERE name = ?', (name,))
        row = cursor.fetchone()
        conn.close()
        return row[0]

    # ========== 子网管理 ==========

    def add_subnet(self, network: str, vlan_id: int = None,
                   site_id: int = None, purpose: str = None,
                   description: str = None, parent_subnet_id: int = None) -> int:
        """添加子网"""
        net = ipaddress.IPv4Network(network, strict=False)

        # 检查是否与现有子网冲突
        if self._check_subnet_overlap(str(net), net.prefixlen):
            raise ValueError(f"子网 {net} 与现有子网冲突")

        conn = sqlite3.connect(self.db_path)
        try:
            cursor = conn.execute(
                '''INSERT INTO subnets (network, prefix_length, parent_subnet_id,
                   site_id, vlan_id, purpose, description)
                   VALUES (?, ?, ?, ?, ?, ?, ?)''',
                (str(net), net.prefixlen, parent_subnet_id,
                 site_id, vlan_id, purpose, description)
            )
            conn.commit()
            subnet_id = cursor.lastrowid
        except sqlite3.IntegrityError:
            cursor = conn.execute(
                'SELECT id FROM subnets WHERE network = ? AND prefix_length = ?',
                (str(net), net.prefixlen)
            )
            row = cursor.fetchone()
            subnet_id = row[0] if row else None
        finally:
            conn.close()

        return subnet_id

    def _check_subnet_overlap(self, network: str, prefix: int) -> bool:
        """检查子网是否与现有子网重叠"""
        new_net = ipaddress.IPv4Network(f"{network}/{prefix}")
        conn = sqlite3.connect(self.db_path)
        cursor = conn.execute('SELECT network, prefix_length FROM subnets')

        for row in cursor.fetchall():
            existing = ipaddress.IPv4Network(f"{row[0]}/{row[1]}")
            if new_net.overlaps(existing):
                conn.close()
                return True
        conn.close()
        return False

    def get_subnet(self, subnet_id: int) -> Optional[Dict]:
        """获取子网信息"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.execute(
            '''SELECT s.id, s.network, s.prefix_length, s.vlan_id, s.purpose,
                      s.description, s.status, sit.name as site_name
               FROM subnets s
               LEFT JOIN sites sit ON s.site_id = sit.id
               WHERE s.id = ?''',
            (subnet_id,)
        )
        row = cursor.fetchone()
        conn.close()

        if row:
            return {
                "id": row[0],
                "network": row[1],
                "prefix_length": row[2],
                "vlan_id": row[3],
                "purpose": row[4],
                "description": row[5],
                "status": row[6],
                "site_name": row[7],
            }
        return None

    # ========== IP 地址分配 ==========

    def allocate_ip(self, subnet_id: int, hostname: str = None,
                    mac_address: str = None, assigned_to: str = None,
                    purpose: str = None, allocation_type: str = "static",
                    skip_first: int = 1, skip_last: int = 1) -> Optional[str]:
        """从子网自动分配一个可用 IP"""
        conn = sqlite3.connect(self.db_path)

        try:
            # 获取子网信息
            cursor = conn.execute(
                'SELECT network, prefix_length FROM subnets WHERE id = ? AND status = ?',
                (subnet_id, 'active')
            )
            row = cursor.fetchone()
            if not row:
                return None

            network = ipaddress.IPv4Network(f"{row[0]}/{row[1]}")

            # 获取已分配的 IP
            cursor = conn.execute(
                'SELECT ip_address FROM ip_allocations WHERE subnet_id = ? AND status = ?',
                (subnet_id, 'active')
            )
            allocated_ips = {row[0] for row in cursor.fetchall()}

            # 保留地址集合
            reserved = {
                str(network.network_address),
                str(network.broadcast_address),
            }

            # 跳过前 skip_first 个和后 skip_last 个地址
            all_hosts = list(network.hosts())
            available_range = all_hosts[skip_first:len(all_hosts)-skip_last] if skip_last > 0 else all_hosts[skip_first:]

            # 遍历可用地址
            for addr in available_range:
                ip_str = str(addr)
                if ip_str not in allocated_ips and ip_str not in reserved:
                    cursor = conn.execute(
                        '''INSERT INTO ip_allocations
                           (ip_address, subnet_id, hostname, mac_address,
                            assigned_to, purpose, allocation_type, status, lease_start)
                           VALUES (?, ?, ?, ?, ?, ?, ?, 'active', datetime('now'))''',
                        (ip_str, subnet_id, hostname, mac_address,
                         assigned_to, purpose, allocation_type)
                    )
                    conn.commit()
                    return ip_str

            return None  # 子网已满

        finally:
            conn.close()

    def release_ip(self, ip_address: str) -> bool:
        """释放 IP 地址"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.execute(
            "UPDATE ip_allocations SET status = 'released', "
            "lease_end = datetime('now'), updated_at = datetime('now') "
            "WHERE ip_address = ? AND status = 'active'",
            (ip_address,)
        )
        changed = cursor.rowcount > 0
        conn.commit()
        conn.close()
        return changed

    def assign_specific_ip(self, ip_address: str, subnet_id: int,
                           hostname: str = None, mac_address: str = None,
                           assigned_to: str = None) -> bool:
        """分配指定 IP 地址"""
        # 验证 IP 是否属于子网
        conn = sqlite3.connect(self.db_path)
        cursor = conn.execute(
            'SELECT network, prefix_length FROM subnets WHERE id = ?',
            (subnet_id,)
        )
        row = cursor.fetchone()

        if not row:
            conn.close()
            return False

        network = ipaddress.IPv4Network(f"{row[0]}/{row[1]}")
        try:
            ip_addr = ipaddress.IPv4Address(ip_address)
        except ValueError:
            conn.close()
            return False

        if ip_addr not in network:
            conn.close()
            return False

        # 检查是否已被分配
        cursor = conn.execute(
            'SELECT id FROM ip_allocations WHERE ip_address = ? AND status = ?',
            (ip_address, 'active')
        )
        if cursor.fetchone():
            conn.close()
            return False

        cursor = conn.execute(
            '''INSERT INTO ip_allocations
               (ip_address, subnet_id, hostname, mac_address, assigned_to,
                allocation_type, status, lease_start)
               VALUES (?, ?, ?, ?, ?, 'static', 'active', datetime('now'))''',
            (ip_address, subnet_id, hostname, mac_address, assigned_to)
        )
        conn.commit()
        conn.close()
        return True

    # ========== IP 地址查询 ==========

    def query_ip(self, ip_address: str) -> Optional[Dict]:
        """查询 IP 地址详细信息"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.execute(
            '''SELECT ia.ip_address, ia.hostname, ia.mac_address,
                      ia.assigned_to, ia.purpose, ia.allocation_type,
                      ia.status, ia.lease_start, ia.lease_end,
                      s.network, s.prefix_length, s.vlan_id,
                      sit.name as site_name
               FROM ip_allocations ia
               JOIN subnets s ON ia.subnet_id = s.id
               LEFT JOIN sites sit ON s.site_id = sit.id
               WHERE ia.ip_address = ? AND ia.status = 'active' ''',
            (ip_address,)
        )
        row = cursor.fetchone()
        conn.close()

        if row:
            return {
                "ip_address": row[0],
                "hostname": row[1],
                "mac_address": row[2],
                "assigned_to": row[3],
                "purpose": row[4],
                "allocation_type": row[5],
                "status": row[6],
                "lease_start": row[7],
                "lease_end": row[8],
                "subnet": f"{row[9]}/{row[10]}",
                "vlan_id": row[11],
                "site_name": row[12],
            }
        return None

    # ========== 子网利用率 ==========

    def get_subnet_utilization(self, subnet_id: int) -> Dict:
        """获取子网利用率"""
        conn = sqlite3.connect(self.db_path)

        cursor = conn.execute(
            'SELECT network, prefix_length FROM subnets WHERE id = ?',
            (subnet_id,)
        )
        row = cursor.fetchone()
        network = ipaddress.IPv4Network(row[0])

        cursor = conn.execute(
            'SELECT COUNT(*) FROM ip_allocations WHERE subnet_id = ? AND status = ?',
            (subnet_id, 'active')
        )
        allocated_count = cursor.fetchone()[0]

        total_hosts = max(2, network.num_addresses - 2)

        conn.close()

        return {
            "subnet": str(network),
            "total_hosts": total_hosts,
            "allocated": allocated_count,
            "available": total_hosts - allocated_count,
            "utilization_percent": round(allocated_count / total_hosts * 100, 1),
        }

    # ========== 冲突检测 ==========

    def detect_ip_conflicts(self) -> List[Dict]:
        """检测 IP 地址冲突"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.execute(
            '''SELECT ip_address, COUNT(*) as cnt
               FROM ip_allocations
               WHERE status = 'active'
               GROUP BY ip_address
               HAVING cnt > 1'''
        )

        conflicts = []
        for row in cursor.fetchall():
            details = conn.execute(
                '''SELECT hostname, assigned_to, lease_start, device_type
                   FROM ip_allocations
                   WHERE ip_address = ? AND status = 'active' ''',
                (row[0],)
            )
            conflicts.append({
                "ip_address": row[0],
                "count": row[1],
                "details": [
                    {
                        "hostname": d[0],
                        "assigned_to": d[1],
                        "lease_start": d[2],
                        "device_type": d[3],
                    }
                    for d in details.fetchall()
                ]
            })
        conn.close()
        return conflicts

    def detect_subnet_overlaps(self) -> List[Dict]:
        """检测子网重叠"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.execute('SELECT id, network, prefix_length FROM subnets')
        subnets = cursor.fetchall()
        conn.close()

        overlaps = []
        for i in range(len(subnets)):
            for j in range(i + 1, len(subnets)):
                net_a = ipaddress.IPv4Network(
                    f"{subnets[i][1]}/{subnets[i][2]}"
                )
                net_b = ipaddress.IPv4Network(
                    f"{subnets[j][1]}/{subnets[j][2]}"
                )
                if net_a.overlaps(net_b):
                    overlaps.append({
                        "subnet_a": f"{subnets[i][1]}/{subnets[i][2]}",
                        "subnet_b": f"{subnets[j][1]}/{subnets[j][2]}",
                        "subnet_a_id": subnets[i][0],
                        "subnet_b_id": subnets[j][0],
                    })

        return overlaps

    # ========== 统计报表 ==========

    def get_allocation_report(self, site_id: int = None) -> Dict:
        """获取地址分配统计报表"""
        conn = sqlite3.connect(self.db_path)

        # 子网统计
        if site_id:
            cursor = conn.execute(
                '''SELECT s.id, s.network, s.prefix_length, s.vlan_id,
                          s.purpose, COUNT(ia.id) as allocated
                   FROM subnets s
                   LEFT JOIN ip_allocations ia
                     ON s.id = ia.subnet_id AND ia.status = 'active'
                   WHERE s.site_id = ? AND s.status = 'active'
                   GROUP BY s.id''',
                (site_id,)
            )
        else:
            cursor = conn.execute(
                '''SELECT s.id, s.network, s.prefix_length, s.vlan_id,
                          s.purpose, COUNT(ia.id) as allocated
                   FROM subnets s
                   LEFT JOIN ip_allocations ia
                     ON s.id = ia.subnet_id AND ia.status = 'active'
                   WHERE s.status = 'active'
                   GROUP BY s.id'''
            )

        subnets = []
        total_allocated = 0
        total_available = 0

        for row in cursor.fetchall():
            subnet_id = row[0]
            network = ipaddress.IPv4Network(f"{row[1]}/{row[2]}")
            allocated = row[5]
            total_hosts = max(2, network.num_addresses - 2)
            available = total_hosts - allocated

            subnets.append({
                "network": str(network),
                "vlan_id": row[3],
                "purpose": row[4],
                "total_hosts": total_hosts,
                "allocated": allocated,
                "available": available,
                "utilization_percent": round(allocated / total_hosts * 100, 1),
            })

            total_allocated += allocated
            total_available += available

        conn.close()

        return {
            "subnets": subnets,
            "total_subnets": len(subnets),
            "total_allocated": total_allocated,
            "total_available": total_available,
            "overall_utilization": round(
                total_allocated / (total_allocated + total_available) * 100, 1
            ) if (total_allocated + total_available) > 0 else 0,
        }

3.2 IPAM REST API 服务

# scripts/ipam_api.py
from ipam_core import IPAMDatabase
from flask import Flask, request, jsonify
import json

app = Flask(__name__)
ipam = IPAMDatabase()

# ========== 站点管理 ==========

@app.route("/api/sites", methods=["POST"])
def create_site():
    """创建站点"""
    data = request.json
    site_id = ipam.add_site(
        name=data["name"],
        region=data.get("region"),
        description=data.get("description")
    )
    return jsonify({"site_id": site_id, "status": "created"})

# ========== 子网管理 ==========

@app.route("/api/subnets", methods=["POST"])
def create_subnet():
    """创建子网"""
    data = request.json
    try:
        subnet_id = ipam.add_subnet(
            network=data["network"],
            vlan_id=data.get("vlan_id"),
            site_id=data.get("site_id"),
            purpose=data.get("purpose"),
            description=data.get("description"),
        )
        return jsonify({"subnet_id": subnet_id, "status": "created"})
    except ValueError as e:
        return jsonify({"error": str(e)}), 409

@app.route("/api/subnets/<int:subnet_id>", methods=["GET"])
def get_subnet(subnet_id):
    """查询子网"""
    subnet = ipam.get_subnet(subnet_id)
    if subnet:
        return jsonify(subnet)
    return jsonify({"error": "Subnet not found"}), 404

@app.route("/api/subnets/<int:subnet_id>/utilization", methods=["GET"])
def subnet_utilization(subnet_id):
    """子网利用率"""
    util = ipam.get_subnet_utilization(subnet_id)
    return jsonify(util)

# ========== IP 地址管理 ==========

@app.route("/api/ip/allocate", methods=["POST"])
def allocate_ip():
    """自动分配 IP 地址"""
    data = request.json
    ip = ipam.allocate_ip(
        subnet_id=data["subnet_id"],
        hostname=data.get("hostname"),
        mac_address=data.get("mac_address"),
        assigned_to=data.get("assigned_to"),
        purpose=data.get("purpose"),
        skip_first=data.get("skip_first", 1),
        skip_last=data.get("skip_last", 1),
    )
    if ip:
        return jsonify({"ip_address": ip, "status": "allocated"})
    return jsonify({"error": "No available IP in subnet"}), 409

@app.route("/api/ip/assign", methods=["POST"])
def assign_specific_ip():
    """分配指定 IP 地址"""
    data = request.json
    success = ipam.assign_specific_ip(
        ip_address=data["ip_address"],
        subnet_id=data["subnet_id"],
        hostname=data.get("hostname"),
        mac_address=data.get("mac_address"),
        assigned_to=data.get("assigned_to"),
    )
    if success:
        return jsonify({"ip_address": data["ip_address"], "status": "assigned"})
    return jsonify({"error": "IP assignment failed"}), 409

@app.route("/api/ip/<ip>", methods=["GET"])
def query_ip(ip):
    """查询 IP 地址"""
    info = ipam.query_ip(ip)
    if info:
        return jsonify(info)
    return jsonify({"error": "IP not found"}), 404

@app.route("/api/ip/release", methods=["POST"])
def release_ip():
    """释放 IP 地址"""
    data = request.json
    success = ipam.release_ip(data["ip_address"])
    if success:
        return jsonify({"status": "released"})
    return jsonify({"error": "IP not found or already released"}), 404

# ========== 检测与报表 ==========

@app.route("/api/conflicts", methods=["GET"])
def check_ip_conflicts():
    """检查 IP 冲突"""
    conflicts = ipam.detect_ip_conflicts()
    return jsonify({
        "conflicts": conflicts,
        "count": len(conflicts),
        "status": "clean" if not conflicts else "conflicts_found"
    })

@app.route("/api/subnet-overlaps", methods=["GET"])
def check_subnet_overlaps():
    """检查子网重叠"""
    overlaps = ipam.detect_subnet_overlaps()
    return jsonify({
        "overlaps": overlaps,
        "count": len(overlaps),
    })

@app.route("/api/report/allocation", methods=["GET"])
def allocation_report():
    """地址分配报表"""
    site_id = request.args.get("site_id", type=int)
    report = ipam.get_allocation_report(site_id)
    return jsonify(report)

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000, debug=True)

3.3 与网络设备集成：自动配置 IP

# scripts/ipam_device_integration.py
from ipam_core import IPAMDatabase
from netmiko import ConnectHandler
import yaml

class IPAMDeviceIntegration:
    """IPAM 与网络设备集成"""

    def __init__(self, ipam_db, inventory_file="devices/inventory.yaml"):
        self.ipam = ipam_db
        with open(inventory_file, "r", encoding="utf-8") as f:
            self.devices = yaml.safe_load(f).get("devices", [])

    def assign_ip_and_configure(self, device_host: str,
                                 subnet_id: int, interface_name: str,
                                 hostname: str = None) -> str:
        """分配 IP 并自动配置到设备"""

        # Step 1: 从 IPAM 分配 IP
        ip_address = self.ipam.allocate_ip(
            subnet_id=subnet_id,
            hostname=hostname,
            assigned_to=f"device:{device_host}/{interface_name}",
            purpose="Device interface",
        )

        if not ip_address:
            raise Exception("子网地址耗尽，无法分配 IP")

        print(f"✓ IPAM 分配 IP: {ip_address}")

        # Step 2: 获取子网信息（计算子网掩码）
        subnet_info = self.ipam.get_subnet(subnet_id)
        prefix = subnet_info["prefix_length"]
        mask = str(ipaddress.IPv4Network(f"0.0.0.0/{prefix}").netmask)

        # Step 3: 查找设备连接信息
        device = self._find_device(device_host)
        if not device:
            raise Exception(f"设备 {device_host} 不在设备清单中")

        # Step 4: 连接设备并配置 IP
        connection = ConnectHandler(**device)

        config_commands = [
            f"interface {interface_name}",
            f"ip address {ip_address} {mask}",
        ]

        if hostname:
            config_commands.append(f"description {hostname}")

        connection.send_config_set(config_commands)
        connection.save_config()
        connection.disconnect()

        print(f"✓ 设备 {device['host']} 接口 {interface_name} "
              f"已配置 IP {ip_address}/{prefix}")

        return ip_address

    def _find_device(self, host_ip: str):
        """查找设备"""
        for device in self.devices:
            if device["host"] == host_ip:
                return device
        return None

    def scan_and_register_arp(self, mac_to_hostname=None):
        """扫描设备 ARP 表并自动注册发现的 IP"""
        registered = 0

        for device in self.devices:
            try:
                connection = ConnectHandler(**device)
                output = connection.send_command("display arp all")
                connection.disconnect()

                # 解析 ARP 表
                lines = output.strip().split("\n")[2:]
                for line in lines:
                    parts = line.split()
                    if len(parts) >= 3:
                        ip = parts[0]
                        mac = parts[1]

                        # 检查是否已在 IPAM 中
                        existing = self.ipam.query_ip(ip)
                        if not existing:
                            hostname = mac_to_hostname.get(mac) if mac_to_hostname else None

                            # 尝试自动注册
                            for subnet_id in self._get_all_subnet_ids():
                                try:
                                    util = self.ipam.get_subnet_utilization(subnet_id)
                                    net = ipaddress.IPv4Network(util["subnet"])
                                    if ipaddress.IPv4Address(ip) in net:
                                        self.ipam.assign_specific_ip(
                                            ip_address=ip,
                                            subnet_id=subnet_id,
                                            hostname=hostname,
                                            mac_address=mac,
                                            assigned_to=f"auto-discovered:{device['host']}",
                                        )
                                        registered += 1
                                        break
                                except Exception:
                                    continue

            except Exception as e:
                print(f"设备 {device['host']} ARP 扫描失败: {e}")

        return registered

    def _get_all_subnet_ids(self):
        """获取所有子网 ID"""
        conn = sqlite3.connect(self.ipam.db_path)
        cursor = conn.execute('SELECT id FROM subnets WHERE status = ?', ('active',))
        ids = [row[0] for row in cursor.fetchall()]
        conn.close()
        return ids

3.4 IPAM 客户端工具

# scripts/ipam_client.py
import requests
import json

class IPAMClient:
    """IPAM REST API 客户端"""

    def __init__(self, base_url="http://localhost:5000/api"):
        self.base_url = base_url
        self.session = requests.Session()

    # 站点
    def create_site(self, name, region=None):
        resp = self.session.post(
            f"{self.base_url}/sites",
            json={"name": name, "region": region}
        )
        return resp.json()

    # 子网
    def create_subnet(self, network, vlan_id=None, site_id=None, purpose=None):
        resp = self.session.post(
            f"{self.base_url}/subnets",
            json={
                "network": network,
                "vlan_id": vlan_id,
                "site_id": site_id,
                "purpose": purpose,
            }
        )
        return resp.json()

    def get_subnet_utilization(self, subnet_id):
        resp = self.session.get(f"{self.base_url}/subnets/{subnet_id}/utilization")
        return resp.json()

    # IP 地址
    def allocate_ip(self, subnet_id, hostname=None, assigned_to=None):
        resp = self.session.post(
            f"{self.base_url}/ip/allocate",
            json={
                "subnet_id": subnet_id,
                "hostname": hostname,
                "assigned_to": assigned_to,
            }
        )
        return resp.json()

    def query_ip(self, ip_address):
        resp = self.session.get(f"{self.base_url}/ip/{ip_address}")
        return resp.json()

    def release_ip(self, ip_address):
        resp = self.session.post(
            f"{self.base_url}/ip/release",
            json={"ip_address": ip_address}
        )
        return resp.json()

    # 检测
    def check_conflicts(self):
        resp = self.session.get(f"{self.base_url}/conflicts")
        return resp.json()

    def check_subnet_overlaps(self):
        resp = self.session.get(f"{self.base_url}/subnet-overlaps")
        return resp.json()

    # 报表
    def get_allocation_report(self):
        resp = self.session.get(f"{self.base_url}/report/allocation")
        return resp.json()


# ========== 使用示例 ==========

if __name__ == "__main__":
    client = IPAMClient()

    # 1. 创建站点
    site = client.create_site("Beijing-DC", region="CN-North")
    print(f"站点创建: {site}")

    # 2. 创建子网
    subnet = client.create_subnet(
        "10.10.10.0/24",
        vlan_id=100,
        site_id=site["site_id"],
        purpose="Server-Network"
    )
    print(f"子网创建: {subnet}")

    # 3. 分配 IP
    for i in range(3):
        ip_result = client.allocate_ip(
            subnet_id=subnet["subnet_id"],
            hostname=f"Server-{i+1:02d}",
            assigned_to=f"Rack-{(i%3)+1}",
        )
        print(f"分配 IP: {ip_result}")

    # 4. 查询 IP
    ip_info = client.query_ip("10.10.10.2")
    print(f"IP 查询: {json.dumps(ip_info, indent=2)}")

    # 5. 子网利用率
    util = client.get_subnet_utilization(subnet["subnet_id"])
    print(f"利用率: {util['utilization_percent']}%")

    # 6. 冲突检测
    conflicts = client.check_conflicts()
    print(f"冲突检测: {conflicts['status']}")

    # 7. 释放 IP
    release = client.release_ip("10.10.10.3")
    print(f"释放结果: {release}")

四、实战配置/案例分析

4.1 案例一：数据中心批量服务器 IP 分配

场景：新上架 200 台服务器，需要自动分配管理口 IP 和业务口 IP。

# scripts/batch_server_ip_assignment.py
from ipam_client import IPAMClient
import json

def batch_allocate_server_ips(client, server_list, mgmt_subnet_id,
                               business_subnet_id):
    """批量分配服务器 IP"""
    results = []

    for server in server_list:
        try:
            # 分配管理口 IP
            mgmt_ip = client.allocate_ip(
                subnet_id=mgmt_subnet_id,
                hostname=f"{server['name']}-mgmt",
                assigned_to=f"{server['name']}/mgmt-interface",
            )

            # 分配业务口 IP
            biz_ip = client.allocate_ip(
                subnet_id=business_subnet_id,
                hostname=f"{server['name']}-biz",
                assigned_to=f"{server['name']}/biz-interface",
            )

            results.append({
                "server": server["name"],
                "rack": server["rack"],
                "mgmt_ip": mgmt_ip["ip_address"],
                "biz_ip": biz_ip["ip_address"],
                "status": "success",
            })

        except Exception as e:
            results.append({
                "server": server["name"],
                "status": "error",
                "error": str(e),
            })

    return results

# 服务器清单
servers = [
    {"name": f"SRV-{i:03d}", "rack": f"R{(i%8)+1}"}
    for i in range(1, 201)
]

client = IPAMClient()
results = batch_allocate_server_ips(client, servers, 1, 2)

# 保存分配结果
with open("server_ip_allocation.json", "w", encoding="utf-8") as f:
    json.dump(results, f, indent=2, ensure_ascii=False)

success_count = sum(1 for r in results if r["status"] == "success")
print(f"IP 分配完成: {success_count}/{len(servers)} 台成功")

4.2 案例二：IP 地址冲突紧急排查

故障场景：用户报告某服务器无法访问，排查发现其 IP 被另一台设备占用。

# scripts/ip_conflict_emergency.py
from ipam_client import IPAMClient

def investigate_ip_conflict(conflict_ip):
    """紧急排查 IP 冲突"""
    client = IPAMClient()

    print(f"=" * 60)
    print(f"IP 冲突排查: {conflict_ip}")
    print(f"=" * 60)

    # 1. 检查 IPAM 中的记录
    ip_info = client.query_ip(conflict_ip)
    if ip_info:
        print(f"\nIPAM 记录:")
        print(f"  主机名: {ip_info.get('hostname', 'N/A')}")
        print(f"  分配给: {ip_info.get('assigned_to', 'N/A')}")
        print(f"  MAC: {ip_info.get('mac_address', 'N/A')}")
        print(f"  子网: {ip_info.get('subnet', 'N/A')}")
        print(f"  分配时间: {ip_info.get('lease_start', 'N/A')}")
    else:
        print(f"\nIPAM 中无此 IP 记录 → 可能是未管理的 IP")

    # 2. 全量冲突检测
    conflicts = client.check_conflicts()
    if conflicts["count"] > 0:
        print(f"\n发现 {conflicts['count']} 处 IP 冲突:")
        for conflict in conflicts["conflicts"]:
            print(f"\n  IP: {conflict['ip_address']}")
            for detail in conflict["details"]:
                print(f"    - {detail.get('hostname', '?')} @ {detail.get('assigned_to', '?')}")
    else:
        print(f"\nIPAM 内部无冲突（冲突可能在 IPAM 外部）")

    # 3. 子网重叠检查
    overlaps = client.check_subnet_overlaps()
    if overlaps["count"] > 0:
        print(f"\n发现 {overlaps['count']} 处子网重叠:")
        for overlap in overlaps["overlaps"]:
            print(f"  {overlap['subnet_a']} 与 {overlap['subnet_b']} 重叠")

    # 4. 建议
    print(f"\n{'='*60}")
    print(f"建议操作:")
    print(f"  1. 在网络上 ping -t {conflict_ip} 确认冲突")
    print(f"  2. 使用 ARP 扫描确定实际占用设备")
    print(f"  3. 释放 IPAM 中的冲突记录")
    print(f"  4. 重新分配正确的 IP")

# 使用
investigate_ip_conflict("10.10.10.50")

4.3 案例三：IP 地址使用趋势分析

# scripts/ip_usage_trend.py
from ipam_client import IPAMClient
import json

def generate_ip_usage_dashboard(client):
    """生成 IP 使用趋势仪表板数据"""
    report = client.get_allocation_report()

    # 按目的分类统计
    purpose_stats = {}
    for subnet in report["subnets"]:
        purpose = subnet.get("purpose", "未分类")
        if purpose not in purpose_stats:
            purpose_stats[purpose] = {"allocated": 0, "total": 0}
        purpose_stats[purpose]["allocated"] += subnet["allocated"]
        purpose_stats[purpose]["total"] += subnet["total_hosts"]

    # 利用率告警（> 80% 的子网）
    high_utilization = [
        s for s in report["subnets"]
        if s["utilization_percent"] > 80
    ]

    # 即将耗尽（> 90%）
    critical_utilization = [
        s for s in report["subnets"]
        if s["utilization_percent"] > 90
    ]

    dashboard = {
        "summary": {
            "total_subnets": report["total_subnets"],
            "total_allocated": report["total_allocated"],
            "total_available": report["total_available"],
            "overall_utilization": report["overall_utilization"],
        },
        "purpose_breakdown": purpose_stats,
        "alerts": {
            "high_utilization_subnets": [
                {"network": s["network"], "utilization": s["utilization_percent"]}
                for s in high_utilization
            ],
            "critical_subnets": [
                {"network": s["network"], "utilization": s["utilization_percent"]}
                for s in critical_utilization
            ],
        },
        "subnet_details": report["subnets"],
    }

    with open("ipam_dashboard.json", "w", encoding="utf-8") as f:
        json.dump(dashboard, f, indent=2, ensure_ascii=False)

    print(f"仪表板数据已生成")
    print(f"  总利用率: {dashboard['summary']['overall_utilization']}%")
    print(f"  高利用率子网: {len(high_utilization)}")
    print(f"  临界子网: {len(critical_utilization)}")

    return dashboard

# 生成
client = IPAMClient()
dashboard = generate_ip_usage_dashboard(client)

五、总结

思考题

1. IPAM 系统如何保证与 DHCP 服务器的数据一致性？当 DHCP 分配的地址与 IPAM 记录不一致时如何处理？
2. 如果 ARP 扫描发现了一个 IP 地址，但无法确定它属于哪个子网（如跨 VLAN 路由），应该如何处理？
3. 在超大规模网络（10000+ IP）中，IPAM 数据库的读写性能如何保障？是否需要分库或缓存策略？

下篇预告：第321篇：全网拓扑自动发现与绘图