摘要

  • 这篇文章专门解决“IP 已经纳管，但状态仍靠人维护”的问题，把状态判断改成可执行的自动治理规则。
  • 方案由两个脚本组成：一个负责事实采集，一个负责状态决策与通知，避免采集逻辑和治理逻辑缠在一起。
  • 它适合作为 1.0 的补强篇，帮助你把 NetBox 从台账系统推进到可持续治理系统。

从“记录资产”到“状态驱动治理”的一次完整落地

一、为什么要做 IP 资产自动化审计 2.0

审计状态

在前一阶段（NetBox 进阶指南 1.0）中，已经实现：

  • NetBox 作为统一 IP 管理平台
  • 基于 Tag 的资产纳入自动化体系
  • 实现“资产可查、状态可看”

但实际运维中发现一个关键问题：

IP 在 NetBox 中存在，并不代表它仍然可用。

1.1 现实问题

  • IP 长期未使用但仍为 active
  • IP 不可达但状态未更新
  • 状态严重依赖人工经验

1.2 2.0 目标

让 IP 状态由“事实驱动”，而不是人工维护

二、最终实现效果

上线后实现：

  • 每日自动 ICMP 探测
  • 7 天不通 → 自动标记 deprecated
  • 30 天不通 → 自动删除
  • 钉钉自动通知
  • 无异常只发送“正常汇总”

核心原则

text
只汇报变化，不制造噪音

三、整体方案拆解

IP 资产自动化治理闭环

治理闭环

系统拆为两个核心脚本：

脚本 职责
ICMP 探测 判断 IP 是否存活
审计脚本 决策状态、删除、告警

四、前置准备

4.1 NetBox 自定义字段

路径：

text
Admin → Custom Fields → IP Address

新增字段：

字段名 类型 用途
last_seen DateTime 最近成功时间
icmp_fail_days Integer 连续失败天数

4.2 运行环境

  • Linux（Ubuntu / CentOS）
  • Python ≥ 3.9
  • 可访问 NetBox API
  • 可 ping 目标 IP

五、目录结构

text
/opt/ip-audit
├── icmp_last_seen_update.py
├── ip_audit_report.py
├── .env
├── logs
└── venv

六、环境准备

bash
cd /opt/ip-audit
python3 -m venv venv
source venv/bin/activate

pip install pynetbox requests python-dotenv tqdm

七、环境变量配置

bash
NB_URL=http://你的-netbox-地址
NB_TOKEN=你的-netbox-token
DINGTALK_WEBHOOK=https://oapi.dingtalk.com/robot/send?access_token=xxxx
DINGTALK_AT_MOBILES=138xxxxxx
AUDIT_TITLE=IP资产自动化审计

八、ICMP 探测脚本（事实采集）

只负责采集，不做决策

python
#!/usr/bin/env python3
import os
import subprocess
from datetime import datetime
from zoneinfo import ZoneInfo
from concurrent.futures import ThreadPoolExecutor, as_completed

import pynetbox
from tqdm import tqdm
from dotenv import load_dotenv

BASE_DIR = os.path.dirname(os.path.abspath(__file__))
load_dotenv(os.path.join(BASE_DIR, ".env"))

NB_URL = os.getenv("NB_URL")
NB_TOKEN = os.getenv("NB_TOKEN")

LAST_SEEN_FIELD = "last_seen"
FAIL_FIELD = "icmp_fail_days"

PING_TIMEOUT = 1
MAX_WORKERS = 80

nb = pynetbox.api(NB_URL, token=NB_TOKEN)
now = datetime.now(ZoneInfo("Asia/Shanghai"))

def icmp_check(ip):
    addr = ip.address.split("/")[0]
    result = subprocess.run(
        ["ping", "-c", "1", "-W", str(PING_TIMEOUT), addr],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )
    return ip, result.returncode == 0

def main():
    ips = list(nb.ipam.ip_addresses.all())

    with ThreadPoolExecutor(max_workers=MAX_WORKERS) as pool:
        futures = [pool.submit(icmp_check, ip) for ip in ips]

        for future in tqdm(as_completed(futures), total=len(futures)):
            ip, alive = future.result()
            cf = ip.custom_fields or {}
            fail_days = int(cf.get(FAIL_FIELD) or 0)

            if alive:
                ip.custom_fields[LAST_SEEN_FIELD] = now.isoformat()
                ip.custom_fields[FAIL_FIELD] = 0
            else:
                ip.custom_fields[FAIL_FIELD] = fail_days + 1

            ip.save()

if __name__ == "__main__":
    main()

九、IP 审计脚本（决策 + 告警）

python
#!/usr/bin/env python3
import os
import json
import requests
from datetime import datetime, timedelta, timezone
from dotenv import load_dotenv
import pynetbox

BASE_DIR = os.path.dirname(os.path.abspath(__file__))
load_dotenv(os.path.join(BASE_DIR, ".env"))

NB_URL = os.getenv("NB_URL")
NB_TOKEN = os.getenv("NB_TOKEN")

DINGTALK_WEBHOOK = os.getenv("DINGTALK_WEBHOOK")
AT_MOBILES = os.getenv("DINGTALK_AT_MOBILES", "").split(",")
TITLE = os.getenv("AUDIT_TITLE", "IP资产自动化审计")

nb = pynetbox.api(NB_URL, token=NB_TOKEN)

FAIL_FIELD = "icmp_fail_days"
LAST_SEEN_FIELD = "last_seen"

now = datetime.now(timezone.utc)
offline_threshold = now - timedelta(days=30)

def send_msg(content):
    at_text = " ".join([f"@{m}" for m in AT_MOBILES if m])
    data = {
        "msgtype": "markdown",
        "markdown": {
            "title": TITLE,
            "text": f"### {TITLE}\n\n{content}\n\n---\n通知人员: {at_text}"
        },
        "at": {"atMobiles": AT_MOBILES}
    }
    requests.post(DINGTALK_WEBHOOK, json=data, timeout=10)

def main():
    ips = nb.ipam.ip_addresses.all()

    offline = []
    deleted = []

    for ip in ips:
        cf = ip.custom_fields or {}
        fail_days = int(cf.get(FAIL_FIELD) or 0)
        last_seen_raw = cf.get(LAST_SEEN_FIELD)

        if fail_days >= 7 and ip.status.value != "deprecated":
            ip.status = "deprecated"
            ip.save()
            offline.append(ip.address)

        if last_seen_raw:
            last_seen = datetime.fromisoformat(last_seen_raw.replace("Z", "+00:00"))
            if last_seen < offline_threshold:
                deleted.append(ip.address)
                ip.delete()

    if not offline and not deleted:
        send_msg(
            "✅ 今日资产审计完毕\n\n"
            "所有已纳入审计的 IP 状态均正常。"
        )
        return

    msg = "📊 审计汇总\n\n"
    if offline:
        msg += f"🚨 新增离线 IP：{len(offline)}\n" + "\n".join(offline) + "\n\n"
    if deleted:
        msg += f"🗑️ 已删除 IP：{len(deleted)}\n" + "\n".join(deleted)

    send_msg(msg)

if __name__ == "__main__":
    main()

十、定时任务

bash
# ICMP 探测
0 3 * * * /opt/ip-audit/venv/bin/python /opt/ip-audit/icmp_last_seen_update.py >> /opt/ip-audit/logs/icmp.log 2>&1

## 审计与告警
0 8 * * * /opt/ip-audit/venv/bin/python /opt/ip-audit/ip_audit_report.py >> /opt/ip-audit/logs/audit.log 2>&1

十一、运行效果说明

场景一：全部正常

text
✅ 今日资产审计完毕
所有 IP 状态正常

场景二：7天不可达

  • 自动标记 deprecated
  • 钉钉告警

场景三：30天未存活

  • 自动删除 IP
  • 自动清理垃圾数据

场景四：恢复在线

  • 自动恢复 active
  • 系统具备自愈能力

十二、整体价值总结

text
从“靠人记” → “靠数据跑”

实现：

  • IP 状态实时可信
  • 自动清理无效资产
  • 降低运维成本
  • 减少告警噪音

十三、后续扩展

未来可扩展：

  • SNMP 存活检测
  • TCP / HTTP 探测
  • 设备联动审计
  • 机柜 / 机房级联动