6 Commits

Author SHA1 Message Date
Your Name 1a3c2d15c8 feat(scripts): add backfill_vendor.py
一次性运维脚本:对 mac_address 有值但 vendor/hostname 为空的 IP,
重新调用 EnhancedScanService.get_mac_vendor (基于 IEEE OUI 数据库)
和 reverse_dns_lookup 回填字段,不需重新扫描。

解决场景:升级前 mac-vendor-lookup 库未装、或扫描时 vendor
字段因 bug 未写入。

- backend/scripts/backfill_vendor.py: 主脚本,支持 dry-run / --with-hostname / --limit
- backend/scripts/README.md: 用法、参数、故障排查
2026-07-20 17:35:04 +08:00
Your Name 845fefd6cf add file 2026-07-20 14:46:47 +08:00
Your Name bca0da7c7d readme 2026-07-20 13:40:05 +08:00
Your Name 9e34f6f0f9 feat(ips): 批量补全厂商功能 + MAC厂商替换为IEEE OUI数据库
前端:
- IPs.vue: 新增「补全厂商」按钮,选中网段后一键补全
- scanApi: 新增 batchFillVendor/batchDiscoverHostname + enable_netbios 参数

后端:
- enhanced_scan.py: 新增 POST /scan/batch/vendor 批量补全厂商端点
- enhanced_scan.py: 新增 POST /scan/batch/hostname 批量发现主机名端点
- scan_tasks.py: 透传 enable_netbios 参数
2026-07-18 10:59:27 +08:00
Your Name ccd7fbe2f0 feat(scan): 增强MAC厂商识别和主机名自动发现
- MAC厂商: 替换硬编码OUI_MAP为mac_vendor_lookup库(IEEE OUI官方数据库)
- 主机名发现新增NetBIOS广播(nmblookup),支持Windows/SMB设备
- 综合扫描支持enable_netbios参数
- 新增NetBIOS主机名发现API端点
- 扫描时自动补全缺失厂商(有MAC无vendor时自动识别)
2026-07-18 10:53:33 +08:00
Your Name bd68abee93 fix(snmp): ARP表缺少发现时间字段,在get_arp_table返回值中添加last_seen和discovered_at 2026-07-18 09:50:20 +08:00
29 changed files with 2319 additions and 591 deletions
+210
View File
@@ -0,0 +1,210 @@
# IPAM 项目开发进度
## ✅ P0 - 核心基础
- ✅ 项目框架搭建 (FastAPI + SQLAlchemy + MySQL)
- ✅ 网段管理 (创建、编辑、删除、查看)
- ✅ IP地址管理 (分页、过滤、搜索、在线状态)
- ✅ ICMP Ping 扫描引擎 (单IP、网段扫描)
- ✅ 数据库模型设计
## ✅ P1 - 增强功能
-**ARP 扫描** - 二层MAC地址发现
-**反向 DNS 解析** - 主机名自动获取
-**MAC 厂商识别** - 内置 IEEE OUI 数据库
-**Celery 异步任务** - 后台扫描任务执行
-**SNMP 设备集成** - SNMPv2c/v3 凭证与设备管理,ARP表自动采集
-**告警与异常检测** - IP冲突、未授权接入、网段耗尽、设备离线等
## ✅ P2 - 前端界面
- ✅ Vue 3 + Element Plus 基础架构
- ✅ 主布局与路由配置 (侧边栏、顶栏、面包屑)
- ✅ 仪表盘页面 (统计卡片、网段使用率)
- ✅ 网段管理页面 (列表、CRUD、扫描)
- ✅ IP地址管理页面 (筛选、搜索、批量扫描、详情编辑)
- ✅ 扫描管理页面 (手动扫描、任务状态)
- ✅ SNMP管理页面 (设备、凭据、ARP表、接口)
- ✅ 告警中心 (告警列表、确认、解决、MAC白名单)
## ✅ P3 - 系统增强
-**JWT 用户认证** - 登录、登出、修改密码
-**权限管理** - 4级角色权限 (super_admin/admin/operator/viewer)
-**操作审计日志** - 记录所有用户操作、资源变更历史
-**报表导出** - CSV导出 (IP、网段、告警、设备、ARP表)、汇总统计报表
-**邮件告警通知** - SMTP邮件发送、告警邮件模板
-**IP历史轨迹记录** - 自动记录IP状态与MAC变化历史
## 🔧 技术栈
### 后端
- FastAPI 0.115
- SQLAlchemy 2.0
- MySQL 8.0
- Celery 5.4 (Redis broker)
- PySNMP 4.4.12
- Passlib + bcrypt (密码加密)
- python-jose (JWT token)
### 前端
- Vue 3.5
- Element Plus 2.9
- Vue Router 4.4
- Pinia 2.4 (状态管理)
- Axios 1.7
---
## 📋 API 端点总览
### 认证 (/api/v1/auth)
- `POST /login` - 用户登录获取Token
- `POST /refresh` - 刷新访问令牌
- `POST /logout` - 登出
- `GET /me` - 获取当前用户信息
- `POST /change-password` - 修改密码
### 用户管理 (/api/v1/auth)
- `GET /users` - 用户列表
- `GET /users/{id}` - 用户详情
- `POST /users` - 创建用户
- `PUT /users/{id}/status` - 修改用户状态
- `POST /users/{id}/unlock` - 解锁用户
- `POST /users/{id}/reset-password` - 重置密码
- `DELETE /users/{id}` - 删除用户
### 网段管理 (/api/v1/networks)
- `GET /networks` - 网段列表
- `GET /networks/{id}` - 网段详情
- `POST /networks` - 创建网段
- `PUT /networks/{id}` - 更新网段
- `DELETE /networks/{id}` - 删除网段
- `GET /networks/{id}/ips` - 获取网段下的IP列表
- `POST /networks/{id}/scan` - 扫描网段
### IP地址管理 (/api/v1/ips)
- `GET /ips` - IP列表(支持筛选、搜索)
- `GET /ips/{id}` - IP详情
- `PUT /ips/{id}` - 更新IP信息
- `POST /ips/{id}/scan` - 扫描单个IP
### 告警管理 (/api/v1/alerts)
- `GET /alerts` - 告警列表
- `POST /alerts/detect/run` - 运行所有检测
- `POST /alerts/{id}/acknowledge` - 确认告警
- `POST /alerts/{id}/resolve` - 解决告警
- `POST /alerts/{id}/ignore` - 忽略告警
- `GET /alerts/statistics` - 告警统计
### MAC白名单 (/api/v1/alerts)
- `GET /whitelist/macs` - 获取白名单
- `POST /whitelist/macs` - 添加MAC到白名单
- `DELETE /whitelist/macs/{id}` - 从白名单移除
### SNMP管理 (/api/v1/snmp)
- `GET /credentials` - 凭据列表
- `POST /credentials` - 创建凭据
- `PUT /credentials/{id}` - 更新凭据
- `DELETE /credentials/{id}` - 删除凭据
- `GET /devices` - 设备列表
- `POST /devices` - 创建设备
- `POST /devices/{id}/test` - 测试连接
- `POST /devices/{id}/poll` - 立即轮询
- `GET /arp-entries` - ARP表
- `GET /statistics` - SNMP统计
### 审计日志 (/api/v1/audit)
- `GET /logs` - 审计日志列表
- `GET /logs/my` - 当前用户操作日志
- `GET /logs/resource/{type}/{id}` - 资源操作历史
- `GET /statistics` - 审计统计
- `DELETE /clean` - 清理旧日志
### 报表导出 (/api/v1/reports)
- `GET /ip-addresses/csv` - 导出IP地址CSV
- `GET /networks/csv` - 导出网段汇总CSV
- `GET /alerts/csv` - 导出告警CSV
- `GET /snmp-devices/csv` - 导出SNMP设备CSV
- `GET /arp-table/csv` - 导出ARP表CSV
- `GET /summary` - IPAM汇总统计报表
---
## 🔐 角色权限矩阵
| 权限\角色 | viewer | operator | admin | super_admin |
|----------|--------|----------|-------|-------------|
| 查看所有数据 | ✅ | ✅ | ✅ | ✅ |
| 编辑IP信息 | ❌ | ✅ | ✅ | ✅ |
| 执行扫描 | ❌ | ✅ | ✅ | ✅ |
| 网段管理 | ❌ | ❌ | ✅ | ✅ |
| SNMP配置 | ❌ | ❌ | ✅ | ✅ |
| 导出报表 | ❌ | ❌ | ✅ | ✅ |
| 用户管理 | ❌ | ❌ | ❌ | ✅ |
| 审计日志查看 | ❌ | ❌ | ✅ | ✅ |
| 系统设置 | ❌ | ❌ | ❌ | ✅ |
---
## 🚀 部署启动
### 后端启动
```bash
cd /root/ipam/backend
source venv/bin/activate
# 启动 API 服务
uvicorn app.main:app --host 0.0.0.0 --port 8008
# 启动 Celery Worker
celery -A app.tasks.celery_app worker --loglevel=info --concurrency=4
```
### 前端启动
```bash
cd /root/ipam/frontend
npm install
npm run dev
```
### 默认管理员
- 用户名: `admin`
- 密码: `admin123`
---
## 📊 数据库表结构
| 表名 | 说明 | 条数 |
|------|------|------|
| networks | 网段表 | 3 |
| ip_addresses | IP地址表 | 1530 |
| scan_tasks | 扫描任务表 | - |
| snmp_credentials | SNMP凭据表 | - |
| network_devices | SNMP设备表 | - |
| arp_entries | ARP记录表 | - |
| mac_address_table | MAC地址表 | - |
| device_interfaces | 设备接口表 | - |
| alerts | 告警表 | - |
| alert_rules | 告警规则表 | - |
| whitelisted_macs | MAC白名单表 | - |
| users | 用户表 | 1 |
| permissions | 权限表 | - |
| role_permissions | 角色权限关联表 | - |
| refresh_tokens | 刷新令牌表 | - |
| audit_logs | 审计日志表 | - |
---
## 🎯 项目完成度
| 阶段 | 进度 | 状态 |
|------|------|------|
| P0 - 核心基础 | 100% | ✅ 完成 |
| P1 - 增强功能 | 100% | ✅ 完成 |
| P2 - 前端界面 | 100% | ✅ 完成 |
| P3 - 系统增强 | 100% | ✅ 完成 |
**总体完成度: 100%** 🎉
+355 -234
View File
@@ -1,277 +1,398 @@
# IPAM 管理系统 - 完整功能
# IPAM 管理系统
企业级 IP 地址管理系统,支持自动化扫描、发现和管理网络 IP 资产。
## ✅ 已完成功能
---
### 🎯 P0 - 核心功能
-**FastAPI 后端框架** - 高性能 RESTful API
-**MySQL 数据库** - 持久化存储
-**Redis 缓存** - 任务队列和缓存
-**网段管理** - CIDR 网段增删改查,自动创建 IP 记录
-**IP 资产台账** - IP 地址全生命周期管理
-**Ping 扫描引擎** - ICMP 在线状态检测
## ✨ 功能特性
### 🚀 P1 - 高级功能
- **ARP 扫描** - 获取 MAC 地址
- **反向 DNS 解析** - 获取主机名
- **MAC 厂商识别** - 内置 OUI 数据库
-**Celery 异步任务** - 后台并发扫描
-**SNMP 设备集成**
- SNMP v2c/v3 支持
- 设备连接测试
- ARP 表自动采集
- 接口信息采集
-**告警与异常检测**
- IP 冲突告警
- 未授权接入检测
- 网段耗尽预警
- 设备离线告警
- 新设备发现告警
- MAC 地址白名单
### 核心管理
- **网段管理** - CIDR 网段创建、分组、自动生成 IP 列表、使用率监控
- **IP 资产台账** - 在线/离线/空闲/保留状态管理、MAC地址、主机名、厂商识别、使用人/业务标注
- **搜索筛选** - 按网段、状态、IP/MAC/主机名快速搜索
### 🎨 P2 - 前端界面
- **Vue 3 + Element Plus** - 现代化 UI 框架
- **仪表盘** - 统计概览、网段使用率、告警统计
- **网段管理** - 网段 CRUD、扫描、导出
- **IP 地址管理** - 筛选、搜索、编辑
- **扫描管理** - 综合扫描、结果导出
-**SNMP 管理** - 设备、凭据、ARP 表
-**告警中心** - 告警列表、统计、白名单
### 智能扫描
- **ICMP Ping 扫描** - 快速检测在线状态
- **ARP 扫描** - 二层网络 MAC 地址发现
- **反向 DNS 解析** - 自动获取主机名
- **MAC 厂商识别** - 内置 IEEE OUI 数据库,识别 3000+ 设备厂商
- **异步任务执行** - Celery 后台并发扫描,不阻塞 Web 服务
## 🚀 快速启动
### SNMP 网络设备集成
- **多版本支持** - SNMPv2c (Community)、SNMPv3 (认证+加密)
- **自动采集** - ARP 表、接口表、设备信息
- **设备状态** - 轮询状态监控、在线/离线检测
### 方式一:使用启动脚本(推荐)
### 告警与异常检测
- **IP 冲突告警** - 同一 IP 对应多个 MAC 地址
- **未授权接入检测** - 新接入设备不在白名单内自动告警
- **网段耗尽预警** - 网段使用率超阈值告警
- **设备离线告警** - 网络设备长时间不在线告警
- **MAC 白名单** - 可信设备免告警
### 用户与安全
- **JWT 身份认证** - Token 登录、自动刷新
- **4 级角色权限** - 只读 / 操作员 / 管理员 / 超级管理员
- **操作审计日志** - 全量操作记录、可追溯查询
- **密码加密存储** - bcrypt 加密
### 报表与导出
- **CSV 格式导出** - IP 地址表、网段汇总、告警记录、SNMP 设备、ARP 表
- **系统汇总报表** - IP 使用率、在线率、告警统计、网段分布
---
## 🚀 快速开始
### 环境要求
- Python 3.11+
- MySQL 8.0+
- Redis 6.0+
- Node.js 18+
### 一键启动所有服务
#### 方式一:使用启动脚本(推荐)
```bash
cd /root/ipam
# 启动所有服务
chmod +x start.sh
./start.sh
```
### 方式二:手动启动
#### 方式二:手动启动
**1. 启动数据库和缓存**
#### 安装依赖环境
```bash
# MySQL 已运行在 3308 端口
# Redis 已运行在 6379 端口
cd /root/ipam/backend
source venv/bin/activate
pip install "pysnmp==4.4.12" "pyasn1<0.5.0" "pysmi<0.4.0"
apt install -y celery
```
**2. 启动后端服务**
##### 1️⃣ 启动后端 API 服务
```bash
cd /root/ipam/backend
source venv/bin/activate
uvicorn app.main:app --host 0.0.0.0 --port 8008
```
**3. 启动 Celery Worker**
##### 2️⃣ 启动 Celery Worker(执行扫描任务)
```bash
cd /root/ipam/backend
source venv/bin/activate
celery -A app.tasks.celery_app worker --loglevel=info --concurrency=4
```
**4. 启动前端服务**
##### 3️⃣ 启动 Celery Beat(定时任务调度器,自动扫描必须启动)
```bash
cd /root/ipam/backend
source venv/bin/activate
celery -A app.tasks.celery_app beat --loglevel=info
```
##### 4️⃣ 启动前端
```bash
cd /root/ipam/frontend
npm install
npm run dev -- --host 0.0.0.0 --port 3000
```
---
## ⏰ 自动扫描配置
### 扫描频率配置
系统默认配置了 **3 个定时扫描任务**
| 任务名称 | 执行频率 | 扫描内容 | 说明 |
|----------|----------|----------|------|
| **每小时全量扫描** | 每 60 分钟 | Ping + ARP + DNS | 快速刷新在线状态 |
| **每日深度扫描** | 每天凌晨 02:00 | Ping + ARP + DNS | 每日完整扫描更新 |
| **统计更新** | 每 15 分钟 | 仅更新统计 | 更新网段使用率 |
### 如何自定义扫描频率
编辑配置文件:`/root/ipam/backend/app/tasks/celery_app.py`
```python
# 定时任务配置 (第 36-55 行)
celery_app.conf.beat_schedule = {
# 每小时执行一次全量扫描
'full-scan-every-hour': {
'task': 'app.tasks.scan_tasks.full_network_scan',
'schedule': 3600.0, # 单位:秒,修改此值调整频率
'args': (True, True, True) # (启用Ping, 启用ARP, 启用DNS)
},
# 每天凌晨2点执行深度扫描
'full-scan-daily': {
'task': 'app.tasks.scan_tasks.full_network_scan',
'schedule': crontab(hour=2, minute=0), # 修改 hour/minute 调整时间
'args': (True, True, True)
},
# 每15分钟更新统计
'update-stats-every-15min': {
'task': 'app.tasks.scan_tasks.update_all_statistics',
'schedule': 900.0, # 单位:秒
},
}
```
**常用配置示例:**
```python
# 每 30 分钟扫描一次
'schedule': 1800.0
# 每天凌晨 3:30 扫描
'schedule': crontab(hour=3, minute=30)
# 工作日上午 9 点扫描
'schedule': crontab(hour=9, minute=0, day_of_week='mon-fri')
# 仅扫描 Ping,不扫描 ARP 和 DNS
'args': (True, False, False)
```
**⚠️ 注意:修改配置后必须重启 Celery Worker 和 Celery Beat 才会生效!**
### 扫描任务包含的内容
每次完整扫描会执行以下操作:
1. **ICMP Ping** - 检测 IP 是否在线
2. **ARP 扫描** - 获取 MAC 地址(同网段有效)
3. **反向 DNS** - 解析主机名
4. **MAC 厂商识别** - 根据 MAC 地址前 3 字节识别厂商
5. **自动更新** - 更新 IP 状态、MAC、主机名、最后发现时间
6. **异常检测** - 触发 IP 冲突、未授权接入等告警检查
### 监控定时任务状态
#### 查看 Celery 活跃任务
```bash
cd /root/ipam/backend
source venv/bin/activate
celery -A app.tasks.celery_app inspect active
```
#### 查看定时任务调度
```bash
celery -A app.tasks.celery_app inspect scheduled
```
#### 查看 Worker 状态
```bash
celery -A app.tasks.celery_app inspect stats
```
---
## 🛠️ 生产环境部署建议
### 使用 Systemd 管理服务(开机自启)
#### 1. 创建 IPAM API 服务
```ini
# /etc/systemd/system/ipam-api.service
[Unit]
Description=IPAM API Service
After=network.target mysql.service redis.service
[Service]
Type=simple
User=root
WorkingDirectory=/root/ipam/backend
Environment="PATH=/root/ipam/backend/venv/bin"
ExecStart=/root/ipam/backend/venv/bin/uvicorn app.main:app --host 0.0.0.0 --port 8008
Restart=always
RestartSec=10
[Install]
WantedBy=multi-user.target
```
#### 2. 创建 Celery Worker 服务
```ini
# /etc/systemd/system/ipam-celery-worker.service
[Unit]
Description=IPAM Celery Worker
After=network.target redis.service
[Service]
Type=simple
User=root
WorkingDirectory=/root/ipam/backend
Environment="PATH=/root/ipam/backend/venv/bin"
ExecStart=/root/ipam/backend/venv/bin/celery -A app.tasks.celery_app worker --loglevel=info --concurrency=4
Restart=always
RestartSec=10
[Install]
WantedBy=multi-user.target
```
#### 3. 创建 Celery Beat 服务
```ini
# /etc/systemd/system/ipam-celery-beat.service
[Unit]
Description=IPAM Celery Beat Scheduler
After=network.target redis.service
[Service]
Type=simple
User=root
WorkingDirectory=/root/ipam/backend
Environment="PATH=/root/ipam/backend/venv/bin"
ExecStart=/root/ipam/backend/venv/bin/celery -A app.tasks.celery_app beat --loglevel=info
Restart=always
RestartSec=10
[Install]
WantedBy=multi-user.target
```
#### 4. 启用并启动服务
```bash
# 重新加载配置
systemctl daemon-reload
# 启用开机自启
systemctl enable ipam-api ipam-celery-worker ipam-celery-beat
# 启动服务
systemctl start ipam-api ipam-celery-worker ipam-celery-beat
# 查看服务状态
systemctl status ipam-api ipam-celery-worker ipam-celery-beat
```
---
## 🌐 访问地址
| 服务 | 地址 | 说明 |
|------|------|------|
| 前端界面 | http://服务器IP:3000 | Web 管理界面 |
| 后端 API | http://服务器IP:8008 | API 服务 |
| API 文档 | http://服务器IP:8008/docs | Swagger 在线文档 |
| MySQL | 服务器IP:3308 | 数据库 |
| Redis | 服务器IP:6379 | 缓存和任务队列 |
| 管理后台 | `http://服务器IP:3000` | Vue 前端界面 |
| API 服务 | `http://服务器IP:8008` | REST API |
| Swagger 文档 | `http://服务器IP:8008/docs` | API 在线测试文档 |
| Redoc 文档 | `http://服务器IP:8008/redoc` | API 文档 |
## 📖 API 使用示例
### 1. 创建网段
```bash
curl -X POST http://localhost:8008/api/v1/networks \
-H "Content-Type: application/json" \
-d '{
"cidr": "192.168.1.0/24",
"name": "办公网络",
"group_name": "office",
"gateway": "192.168.1.1"
}'
```
### 2. 触发扫描
```bash
curl -X POST http://localhost:8008/api/v1/scan/comprehensive/1
```
### 3. 查看统计
```bash
curl http://localhost:8008/api/v1/networks/stats
```
## 🏗️ 项目结构
```
/root/ipam/
├── backend/ # 后端代码
│ ├── app/
│ │ ├── api/ # API 路由
│ │ │ └── v1/
│ │ │ ├── networks.py # 网段管理
│ │ │ ├── ips.py # IP管理
│ │ │ ├── scan.py # 扫描管理
│ │ │ ├── enhanced_scan.py # 增强扫描
│ │ │ ├── snmp.py # SNMP管理
│ │ │ └── alerts.py # 告警管理
│ │ ├── core/ # 核心配置
│ │ │ ├── config.py # 配置文件
│ │ │ └── database.py # 数据库连接
│ │ ├── models/ # 数据模型
│ │ │ ├── network.py # 网段/IP/扫描任务
│ │ │ ├── snmp.py # SNMP设备/凭据
│ │ │ └── alert.py # 告警模型
│ │ ├── schemas/ # Pydantic 验证模型
│ │ │ └── network.py
│ │ ├── services/ # 业务逻辑
│ │ │ ├── network_service.py # 网段服务
│ │ │ ├── scan_service.py # Ping扫描服务
│ │ │ ├── enhanced_scan_service.py # ARP/DNS扫描服务
│ │ │ ├── snmp_service.py # SNMP服务
│ │ │ └── alert_service.py # 告警服务
│ │ ├── tasks/ # Celery 任务
│ │ │ ├── celery_app.py # Celery 应用
│ │ │ └── scan_tasks.py # 扫描任务
│ │ └── main.py # FastAPI 应用入口
│ ├── requirements.txt # Python 依赖
│ └── venv/ # Python 虚拟环境
├── frontend/ # 前端代码
│ ├── src/
│ │ ├── views/ # 页面组件
│ │ │ ├── Dashboard.vue # 仪表盘
│ │ │ ├── Networks.vue # 网段管理
│ │ │ ├── IPs.vue # IP地址
│ │ │ ├── Scan.vue # 扫描管理
│ │ │ ├── SNMP.vue # SNMP管理
│ │ │ └── Alerts.vue # 告警中心
│ │ ├── layout/ # 布局组件
│ │ │ └── MainLayout.vue # 主布局
│ │ ├── router/ # 路由配置
│ │ │ └── index.js
│ │ ├── api/ # API 客户端
│ │ │ └── index.js
│ │ ├── utils/ # 工具函数
│ │ │ └── api.js
│ │ ├── App.vue # 根组件
│ │ └── main.js # 应用入口
│ ├── package.json
│ └── vite.config.js # Vite 配置
├── docs/ # 文档
├── docker-compose.yml # Docker 配置
├── start.sh # 启动脚本
└── README.md
```
## 🔧 技术栈
### 后端
- **Web 框架**: FastAPI 0.110
- **ORM**: SQLAlchemy 2.0
- **数据库**: MySQL 8.0
- **缓存/队列**: Redis 7.0
- **异步任务**: Celery 5.3
- **SNMP**: PySNMP 4.4.12
- **网络探测**: Scapy 2.5.0
- **HTTP 客户端**: httpx
### 前端
- **框架**: Vue 3.5
- **UI 组件**: Element Plus 2.14
- **路由**: Vue Router 4.6
- **状态管理**: Pinia 4.0
- **HTTP 客户端**: Axios 1.18
- **构建工具**: Vite 5.0
## 📊 数据库表
| 表名 | 说明 |
|------|------|
| networks | 网段信息 |
| ip_addresses | IP 地址台账 |
| scan_tasks | 扫描任务记录 |
| snmp_credentials | SNMP 凭据配置 |
| network_devices | 网络设备列表 |
| arp_entries | ARP 表记录 |
| mac_address_table | MAC 地址表 |
| device_interfaces | 设备接口信息 |
| alerts | 告警记录 |
| alert_rules | 告警规则配置 |
| whitelisted_macs | MAC 地址白名单 |
## 🎯 核心特性
### 1. 智能扫描
- **ICMP Ping**: 快速检测在线状态
- **ARP 扫描**: 二层网络 MAC 发现
- **反向 DNS**: 自动解析主机名
- **并发扫描**: Celery 异步执行,不阻塞 Web 服务
### 2. SNMP 设备管理
- 支持 SNMP v2c/v3
- 自动采集 ARP 表、接口信息
- 设备连通性测试
- 轮询状态监控
### 3. 智能告警
- IP 冲突自动检测
- 未授权设备接入告警
- 网段使用率预警
- MAC 白名单机制
### 4. 厂商识别
- 内置 IEEE OUI 数据库
- 自动识别 3000+ 设备厂商
- 支持 Cisco、H3C、华为、华三等主流厂商
## 📝 开发计划
### 待实现功能 (P3)
- [ ] 历史趋势图表
- [ ] 报告导出
- [ ] 用户权限管理
- [ ] 邮件/钉钉/企业微信通知
- [ ] VLAN 自动发现
- [ ] 拓扑图自动生成
- [ ] 端口控制(关闭/开启)
- [ ] 资产变更审计日志
## 🔐 安全建议
1. **修改默认密码**
- MySQL root 密码
- Redis 密码配置
2. **配置防火墙**
- 仅开放必要端口:3000(前端)、8008(API)
- 限制 3308(MySQL)、6379(Redis) 访问来源
3. **HTTPS**
- 生产环境建议配置 Nginx 反向代理 + SSL
4. **访问控制**
- 配置 API 认证 (JWT)
- 限制管理后台访问来源
## 📞 技术支持
如有问题或建议,请查看 API 文档:
- Swagger UI: http://服务器IP:8008/docs
- ReDoc: http://服务器IP:8008/redoc
**默认管理员账号:**
- 用户名: `admin`
- 密码: `admin123`
---
**版本**: 1.0.0
**最后更新**: 2026-07-17
## 📊 技术栈
| 组件 | 技术选型 | 版本 |
|------|----------|------|
| Web 框架 | FastAPI | 0.115 |
| ORM | SQLAlchemy | 2.0 |
| 数据库 | MySQL | 8.0 |
| 异步任务 | Celery + Redis | 5.4 |
| SNMP | PySNMP | 4.4.12 |
| 密码加密 | Passlib + bcrypt | - |
| JWT 认证 | python-jose | - |
| 前端框架 | Vue 3 + Element Plus | 3.5 / 2.9 |
| 路由 | Vue Router | 4.4 |
| 状态管理 | Pinia | 2.4 |
| HTTP 客户端 | Axios | 1.7 |
---
## 🔐 角色权限矩阵
| 操作 | Viewer (只读) | Operator (操作员) | Admin (管理员) | Super Admin (超级管理员) |
|------|---------------|------------------|----------------|---------------------|
| 查看所有数据 | ✅ | ✅ | ✅ | ✅ |
| 编辑 IP 信息 | ❌ | ✅ | ✅ | ✅ |
| 执行扫描 | ❌ | ✅ | ✅ | ✅ |
| 确认/解决告警 | ❌ | ✅ | ✅ | ✅ |
| 网段管理 | ❌ | ❌ | ✅ | ✅ |
| SNMP 设备/凭据 | ❌ | ❌ | ✅ | ✅ |
| MAC 白名单 | ❌ | ❌ | ✅ | ✅ |
| 导出报表 | ❌ | ❌ | ✅ | ✅ |
| 查看审计日志 | ❌ | ❌ | ✅ | ✅ |
| 用户管理 | ❌ | ❌ | ❌ | ✅ |
| 系统设置 | ❌ | ❌ | ❌ | ✅ |
---
## 📋 API 端点总览
| 模块 | 端点数量 | 主要功能 |
|------|----------|----------|
| 认证 | 10 | 登录、登出、Token刷新、用户管理 |
| 网段管理 | 7 | CRUD、扫描、IP列表、统计 |
| IP 地址 | 5 | 列表、详情、编辑、扫描 |
| 扫描 | 8 | 单IP、网段、异步任务、状态查询 |
| SNMP | 13 | 凭据、设备、轮询、ARP表、接口 |
| 告警 | 11 | 列表、确认/解决/忽略、检测、白名单 |
| 审计日志 | 5 | 日志查询、用户日志、资源历史、统计、清理 |
| 报表 | 6 | CSV导出、汇总统计 |
---
## 💡 常见问题
### Q: 为什么 IP 扫描后 MAC 地址都是空的?
A: ARP 扫描只能获取**同网段**设备的 MAC 地址。跨网段需要配置 SNMP 采集核心交换机的 ARP 表。
### Q: 定时扫描没有执行?
A: 请检查 **Celery Beat** 是否正常启动,Celery Worker 是否正常运行,Redis 连接是否正常。
### Q: 如何临时禁用自动扫描?
A: 停止 Celery Beat 服务即可,不会影响手动触发扫描。
### Q: 如何调整扫描并发数?
A: 启动 Celery Worker 时调整 `--concurrency` 参数:
```bash
celery -A app.tasks.celery_app worker --concurrency=8 # 8并发
```
### Q: 扫描时间太长怎么办?
A: 1. 增加并发数 2. 关闭 DNS 解析(较慢)3. 分网段分时段扫描
---
## 📁 项目结构
```
/root/ipam/
├── backend/
│ ├── app/
│ │ ├── api/v1/ # API 路由
│ │ ├── core/ # 配置、数据库、安全
│ │ ├── models/ # 数据模型
│ │ ├── schemas/ # Pydantic 验证模型
│ │ ├── services/ # 业务逻辑层
│ │ ├── tasks/ # Celery 异步任务
│ │ └── main.py # 应用入口
│ └── venv/ # Python 虚拟环境
├── frontend/
│ ├── src/
│ │ ├── views/ # 页面组件
│ │ ├── components/ # 通用组件
│ │ ├── router/ # 路由配置
│ │ └── main.js # 应用入口
│ └── node_modules/
├── start.sh # 一键启动脚本
├── stop.sh # 停止服务脚本
├── DEV_PROGRESS.md # 开发进度文档
└── README.md # 本文件
```
---
## 📞 支持
如遇问题,请查看:
1. API 文档:`http://<服务器IP>:8008/docs`
2. 开发进度:`DEV_PROGRESS.md`
3. Celery 日志:查看 Worker 和 Beat 终端输出
---
## 📝 License
MIT License
Binary file not shown.
+2 -1
View File
@@ -1,5 +1,5 @@
from fastapi import APIRouter
from app.api.v1 import networks, ips, scan, enhanced_scan, async_scan, snmp, alerts, auth, audit
from app.api.v1 import networks, ips, scan, enhanced_scan, async_scan, snmp, alerts, auth, audit, reports
api_router = APIRouter(prefix="/api/v1")
@@ -12,3 +12,4 @@ api_router.include_router(snmp.router)
api_router.include_router(alerts.router)
api_router.include_router(auth.router)
api_router.include_router(audit.router)
api_router.include_router(reports.router)
Binary file not shown.
Binary file not shown.
Binary file not shown.
+128 -127
View File
@@ -1,164 +1,165 @@
from fastapi import APIRouter, Depends, HTTPException, Query
from fastapi.responses import StreamingResponse
from fastapi import APIRouter, Depends, Query
from sqlalchemy.orm import Session
from typing import Optional
from datetime import datetime
import csv
import io
from app.core.database import get_db
from app.core.security import get_current_user
from app.core.security import get_current_user, require_permission
from app.models.auth import User
from app.models.audit import AuditAction, AuditResource
from app.services.audit_service import AuditService
router = APIRouter(
prefix="/audit",
tags=["审计日志"],
dependencies=[Depends(get_current_user)],
)
router = APIRouter(prefix="/audit", tags=["审计日志"])
@router.get("/logs", summary="查询审计日志")
@router.get("/logs", summary="获取审计日志列表")
def get_audit_logs(
user_id: Optional[int] = None,
username: Optional[str] = None,
action: Optional[str] = None,
resource_type: Optional[str] = None,
status: Optional[str] = None,
action: Optional[AuditAction] = None,
resource: Optional[AuditResource] = None,
success: Optional[bool] = None,
start_time: Optional[datetime] = None,
end_time: Optional[datetime] = None,
skip: int = Query(0, ge=0),
limit: int = Query(50, ge=1, le=500),
# 当前用户必须是 admin 或 super_admin 才能看审计
current_user: User = Depends(get_current_user),
db: Session = Depends(get_db),
limit: int = Query(100, ge=1, le=500),
current_user: User = Depends(require_permission("audit:view")),
db: Session = Depends(get_db)
):
# 角色检查:super_admin / admin 可以看所有人的,operator/viewer 只能看自己的
if current_user.role not in ("super_admin", "admin"):
user_id = current_user.id
"""
获取审计日志列表(需要管理员权限)
- **user_id**: 按用户ID过滤
- **action**: 按操作类型过滤 (create/read/update/delete/login/logout/scan/export)
- **resource**: 按资源类型过滤
- **success**: 按操作结果过滤
- **start_time**: 开始时间
- **end_time**: 结束时间
"""
total, items = AuditService.get_logs(
db,
user_id=user_id,
username=username,
action=action,
resource_type=resource_type,
status=status,
resource=resource,
success=success,
start_time=start_time,
end_time=end_time,
skip=skip,
limit=limit,
limit=limit
)
serialized = []
for log in items:
serialized.append({
"id": log.id,
"user_id": log.user_id,
"username": log.username,
"action": log.action,
"resource_type": log.resource_type,
"resource_id": log.resource_id,
"resource_name": log.resource_name,
"method": log.method,
"path": log.path,
"ip_address": log.ip_address,
"user_agent": log.user_agent,
"status": log.status,
"detail": log.detail,
"created_at": log.created_at.isoformat() if log.created_at else None,
# 格式化返回
result_items = []
for item in items:
changed = item.changed_fields.split(',') if item.changed_fields else []
result_items.append({
"id": item.id,
"user_id": item.user_id,
"username": item.username,
"real_name": item.real_name,
"user_ip": item.user_ip,
"action": item.action.value,
"resource": item.resource.value,
"resource_id": item.resource_id,
"description": item.description,
"changed_fields": changed,
"request_method": item.request_method,
"request_path": item.request_path,
"success": bool(item.success),
"error_message": item.error_message,
"created_at": item.created_at
})
return {"total": total, "items": serialized}
return {"total": total, "items": result_items}
@router.get("/logs/export", summary="导出审计日志为 CSV")
def export_audit_logs(
user_id: Optional[int] = None,
username: Optional[str] = None,
action: Optional[str] = None,
resource_type: Optional[str] = None,
status: Optional[str] = None,
start_time: Optional[datetime] = None,
end_time: Optional[datetime] = None,
@router.get("/logs/my", summary="获取当前用户的操作日志")
def get_my_logs(
skip: int = Query(0, ge=0),
limit: int = Query(50, ge=1, le=200),
current_user: User = Depends(get_current_user),
db: Session = Depends(get_db),
db: Session = Depends(get_db)
):
"""导出 CSV 流。admin/super_admin 看全部;operator/viewer 仅能导出自己的。"""
if current_user.role not in ("super_admin", "admin"):
user_id = current_user.id
# 导出上限 50000 行,避免一次拉太多 OOM
"""获取当前登录用户的操作日志"""
total, items = AuditService.get_logs(
db,
user_id=user_id,
username=username,
action=action,
resource_type=resource_type,
status=status,
start_time=start_time,
end_time=end_time,
skip=0,
limit=50000,
)
output = io.StringIO()
# 写入 UTF-8 BOM 让 Excel 直接打开不乱码
output.write("\ufeff")
writer = csv.writer(output)
writer.writerow([
"ID", "时间", "用户", "操作", "资源类型", "资源ID", "资源名称",
"HTTP方法", "路径", "客户端IP", "状态", "详情"
])
for log in items:
writer.writerow([
log.id,
log.created_at.isoformat() if log.created_at else "",
log.username or "",
log.action or "",
log.resource_type or "",
log.resource_id or "",
log.resource_name or "",
log.method or "",
log.path or "",
log.ip_address or "",
log.status or "",
(log.detail or "").replace("\n", " ")[:500],
])
output.seek(0)
filename = f"audit_logs_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
return StreamingResponse(
iter([output.getvalue()]),
media_type="text/csv; charset=utf-8",
headers={"Content-Disposition": f'attachment; filename="{filename}"'},
user_id=current_user.id,
skip=skip,
limit=limit
)
result_items = []
for item in items:
result_items.append({
"id": item.id,
"action": item.action.value,
"resource": item.resource.value,
"resource_id": item.resource_id,
"description": item.description,
"success": bool(item.success),
"created_at": item.created_at
})
return {"total": total, "items": result_items}
@router.get("/stats", summary="审计日志统计")
def get_audit_stats(
current_user: User = Depends(get_current_user),
db: Session = Depends(get_db),
@router.get("/logs/resource/{resource_type}/{resource_id}", summary="获取资源操作历史")
def get_resource_logs(
resource_type: str,
resource_id: str,
limit: int = Query(50, ge=1, le=200),
current_user: User = Depends(require_permission("audit:view")),
db: Session = Depends(get_db)
):
if current_user.role not in ("super_admin", "admin"):
# operator/viewer 只能看自己的简单计数
from app.models.audit import AuditLog
from sqlalchemy import func
my_count = db.query(func.count(AuditLog.id)).filter(AuditLog.user_id == current_user.id).scalar()
return {"recent_24h": 0, "by_action_24h": {}, "my_total": int(my_count or 0)}
return AuditService.get_stats(db)
"""获取指定资源的操作历史"""
try:
resource_enum = AuditResource(resource_type)
except ValueError:
return {"total": 0, "items": [], "error": "无效的资源类型"}
items = AuditService.get_resource_logs(db, resource_enum, resource_id, limit=limit)
result_items = []
for item in items:
changed = item.changed_fields.split(',') if item.changed_fields else []
result_items.append({
"id": item.id,
"username": item.username,
"real_name": item.real_name,
"action": item.action.value,
"description": item.description,
"changed_fields": changed,
"success": bool(item.success),
"created_at": item.created_at
})
return {"total": len(items), "items": result_items}
@router.get("/actions", summary="获取所有审计操作类型(用于前端过滤下拉框)")
def get_action_types():
"""硬编码返回常用 action 列表,避免每次都查 DB"""
from app.services.audit_service import AuditAction
actions = []
for name in dir(AuditAction):
if name.startswith("_") or name.isupper() and not name.startswith("__"):
continue
val = getattr(AuditAction, name, None)
if isinstance(val, str):
actions.append({"code": val, "name": name})
return actions
@router.get("/statistics", summary="获取审计统计信息")
def get_audit_statistics(
days: int = Query(30, ge=1, le=365),
current_user: User = Depends(require_permission("audit:view")),
db: Session = Depends(get_db)
):
"""获取审计统计信息"""
stats = AuditService.get_statistics(db, days=days)
return stats
@router.delete("/clean", summary="清理旧日志")
def clean_old_logs(
days: int = Query(90, ge=7, le=3650, description="保留天数"),
current_user: User = Depends(require_permission("system:admin")),
db: Session = Depends(get_db)
):
"""
清理指定天数之前的审计日志(需要超级管理员权限)
- **days**: 保留日志的天数,默认90天
"""
deleted_count = AuditService.clean_old_logs(db, days=days)
return {
"message": f"清理完成,已删除 {deleted_count} 条旧日志",
"deleted_count": deleted_count,
"kept_days": days
}
+93 -1
View File
@@ -24,6 +24,7 @@ def comprehensive_scan(
enable_ping: bool = True,
enable_arp: bool = True,
enable_dns: bool = True,
enable_netbios: bool = True,
timeout: int = 2,
db: Session = Depends(get_db)
):
@@ -32,7 +33,8 @@ def comprehensive_scan(
- Ping 扫描检测在线状态
- ARP 扫描获取MAC地址
- 反向DNS解析获取主机名
- MAC厂商识别
- NetBIOS 广播获取主机名(Windows/SMB 设备)
- MAC 厂商识别(IEEE OUI 数据库)
"""
network = NetworkService.get_by_id(db, network_id)
if not network:
@@ -51,6 +53,7 @@ def comprehensive_scan(
enable_ping=enable_ping,
enable_arp=enable_arp,
enable_dns=enable_dns,
enable_netbios=enable_netbios,
timeout=timeout
)
@@ -90,6 +93,7 @@ def comprehensive_scan_ip(
enable_ping: bool = True,
enable_arp: bool = True,
enable_dns: bool = True,
enable_netbios: bool = True,
timeout: int = 2,
db: Session = Depends(get_db)
):
@@ -101,6 +105,7 @@ def comprehensive_scan_ip(
enable_ping=enable_ping,
enable_arp=enable_arp,
enable_dns=enable_dns,
enable_netbios=enable_netbios,
timeout=timeout
)
@@ -140,6 +145,93 @@ def reverse_dns(ip_address: str, timeout: int = 2):
"hostname": hostname or "Unknown"
}
@router.post("/netbios/lookup/{ip_address}", summary="NetBIOS 主机名发现")
def netbios_lookup(ip_address: str, timeout: int = 3):
"""
通过 NetBIOS 广播协议获取主机名(Windows / SMB 设备)
"""
hostname = EnhancedScanService.netbios_lookup(ip_address, timeout)
return {
"ip_address": ip_address,
"hostname": hostname or "Unknown"
}
@router.post("/batch/vendor", summary="批量补全厂商信息")
def batch_fill_vendor(
network_id: Optional[int] = None,
db: Session = Depends(get_db)
):
"""
对已有 MAC 地址但缺少厂商信息的 IP 批量补全厂商(使用 IEEE OUI 数据库)
"""
from app.models.network import IPAddress as IPAddressModel
query = db.query(IPAddressModel).filter(
IPAddressModel.mac_address.isnot(None),
IPAddressModel.vendor.is_(None)
)
if network_id:
query = query.filter(IPAddressModel.network_id == network_id)
items = query.all()
updated_count = 0
for ip in items:
vendor = EnhancedScanService.get_mac_vendor(ip.mac_address)
if vendor:
ip.vendor = vendor
updated_count += 1
if updated_count > 0:
db.commit()
return {
"message": f"已补全 {updated_count} 条厂商信息",
"checked": len(items),
"updated": updated_count
}
@router.post("/batch/hostname", summary="批量发现主机名")
def batch_discover_hostname(
network_id: Optional[int] = None,
enable_dns: bool = True,
enable_netbios: bool = True,
db: Session = Depends(get_db)
):
"""
对已有 MAC 地址但缺少主机名的 IP 批量执行主机名发现(DNS PTR + NetBIOS
"""
from app.models.network import IPAddress as IPAddressModel
query = db.query(IPAddressModel).filter(
IPAddressModel.mac_address.isnot(None),
IPAddressModel.hostname.is_(None)
)
if network_id:
query = query.filter(IPAddressModel.network_id == network_id)
items = query.all()
updated_count = 0
for ip in items:
hostname = None
if enable_dns:
hostname = EnhancedScanService.reverse_dns_lookup(ip.ip_address, timeout=2)
if not hostname and enable_netbios:
hostname = EnhancedScanService.netbios_lookup(ip.ip_address, timeout=3)
if hostname:
ip.hostname = hostname
updated_count += 1
if updated_count > 0:
db.commit()
return {
"message": f"已发现 {updated_count} 条主机名",
"checked": len(items),
"updated": updated_count
}
@router.get("/online/ips", summary="获取所有在线IP列表")
def get_online_ips(
+115
View File
@@ -0,0 +1,115 @@
from fastapi import APIRouter, Depends, Query, Response
from sqlalchemy.orm import Session
from typing import Optional
from app.core.database import get_db
from app.core.security import require_permission
from app.models.network import IPStatus
from app.services.report_service import ReportExportService
router = APIRouter(prefix="/reports", tags=["报表导出"])
@router.get("/ip-addresses/csv", summary="导出IP地址CSV")
def export_ip_addresses_csv(
network_id: Optional[int] = None,
status: Optional[IPStatus] = None,
online_only: bool = Query(False, description="仅导出在线IP"),
db: Session = Depends(get_db),
current_user = Depends(require_permission("report:export"))
):
"""导出IP地址表CSV"""
csv_content, filename = ReportExportService.export_ip_addresses_csv(
db,
network_id=network_id,
status=status,
include_online_only=online_only
)
return Response(
content=csv_content,
media_type="text/csv; charset=utf-8-sig",
headers={
"Content-Disposition": f"attachment; filename*=UTF-8''{filename}"
}
)
@router.get("/networks/csv", summary="导出网段汇总CSV")
def export_networks_csv(
db: Session = Depends(get_db),
current_user = Depends(require_permission("report:export"))
):
"""导出网段汇总CSV"""
csv_content, filename = ReportExportService.export_networks_csv(db)
return Response(
content=csv_content,
media_type="text/csv; charset=utf-8-sig",
headers={
"Content-Disposition": f"attachment; filename*=UTF-8''{filename}"
}
)
@router.get("/alerts/csv", summary="导出告警CSV")
def export_alerts_csv(
status: Optional[str] = None,
db: Session = Depends(get_db),
current_user = Depends(require_permission("report:export"))
):
"""导出告警CSV"""
csv_content, filename = ReportExportService.export_alerts_csv(db, status=status)
return Response(
content=csv_content,
media_type="text/csv; charset=utf-8-sig",
headers={
"Content-Disposition": f"attachment; filename*=UTF-8''{filename}"
}
)
@router.get("/snmp-devices/csv", summary="导出SNMP设备CSV")
def export_snmp_devices_csv(
db: Session = Depends(get_db),
current_user = Depends(require_permission("report:export"))
):
"""导出SNMP设备CSV"""
csv_content, filename = ReportExportService.export_snmp_devices_csv(db)
return Response(
content=csv_content,
media_type="text/csv; charset=utf-8-sig",
headers={
"Content-Disposition": f"attachment; filename*=UTF-8''{filename}"
}
)
@router.get("/arp-table/csv", summary="导出ARP表CSV")
def export_arp_table_csv(
device_id: Optional[int] = None,
db: Session = Depends(get_db),
current_user = Depends(require_permission("report:export"))
):
"""导出ARP表CSV"""
csv_content, filename = ReportExportService.export_arp_table_csv(db, device_id=device_id)
return Response(
content=csv_content,
media_type="text/csv; charset=utf-8-sig",
headers={
"Content-Disposition": f"attachment; filename*=UTF-8''{filename}"
}
)
@router.get("/summary", summary="获取IPAM系统汇总报表")
def get_ipam_summary_report(
db: Session = Depends(get_db),
current_user = Depends(require_permission("report:view"))
):
"""获取IPAM系统汇总统计报表"""
summary = ReportExportService.get_ipam_summary(db)
return summary
+5 -5
View File
@@ -285,7 +285,7 @@ def get_network_device(device_id: int, db: Session = Depends(get_db)):
def create_network_device(
name: str,
ip_address: str,
credential_id: Optional[int] = None,
snmp_credential_id: Optional[int] = None, # 前端字段名(与 ORM 列名一致)
port: int = 161,
device_type: str = "other",
description: Optional[str] = None,
@@ -305,7 +305,7 @@ def create_network_device(
ip_address=ip_address,
port=port,
device_type=device_type,
snmp_credential_id=credential_id,
snmp_credential_id=snmp_credential_id,
description=description,
location=location,
is_active=True
@@ -337,7 +337,7 @@ def update_network_device(
device_id: int,
name: Optional[str] = None,
ip_address: Optional[str] = None,
credential_id: Optional[int] = None,
snmp_credential_id: Optional[int] = None,
port: Optional[int] = None,
device_type: Optional[str] = None,
description: Optional[str] = None,
@@ -363,8 +363,8 @@ def update_network_device(
device.name = name
if ip_address:
device.ip_address = ip_address
if credential_id is not None:
device.snmp_credential_id = credential_id
if snmp_credential_id is not None:
device.snmp_credential_id = snmp_credential_id
if port:
device.port = port
if device_type:
+6 -6
View File
@@ -19,17 +19,17 @@ async def lifespan(app: FastAPI):
"""应用生命周期管理"""
# 启动时创建数据库表
Base.metadata.create_all(bind=engine)
# 初始化默认管理员账户 + 默认权限
# 初始化默认管理员账户
from app.services.user_service import UserService
db = SessionLocal()
try:
UserService.init_default_admin(db)
UserService.init_default_permissions(db)
print("✅ 系统初始化完成,默认管理员账户: admin / admin123")
print("✅ IPAM System initialized successfully!")
print(" Default admin: admin / admin123")
finally:
db.close()
yield
# 关闭时清理
@@ -37,7 +37,7 @@ async def lifespan(app: FastAPI):
app = FastAPI(
title=settings.APP_NAME,
version=settings.APP_VERSION,
description="企业级IP地址管理系统",
description="Enterprise IP Address Management System",
lifespan=lifespan
)
+85 -28
View File
@@ -1,37 +1,94 @@
from sqlalchemy import Column, Integer, String, Text, DateTime, ForeignKey, Index
from sqlalchemy import Column, Integer, String, Text, DateTime, Enum
from sqlalchemy.sql import func
from app.core.database import Base
import enum
class AuditAction(str, enum.Enum):
"""操作类型枚举"""
CREATE = "create"
READ = "read"
UPDATE = "update"
DELETE = "delete"
LOGIN = "login"
LOGOUT = "logout"
SCAN = "scan"
IMPORT = "import"
EXPORT = "export"
OTHER = "other"
# ===== 兼容层:老 AuditAction 字符串字面量别名,值与新枚举对应 =====
# 旧代码(以及当前 AuditService._ACTION_MAP)里硬编码引用这些枚举属性名
NETWORK_CREATE = "create"
NETWORK_UPDATE = "update"
NETWORK_DELETE = "delete"
IP_UPDATE = "update"
IP_RESERVE = "update"
IP_UNRESERVE = "update"
SCAN_TRIGGER = "scan"
SNMP_CRED_CREATE = "create"
SNMP_CRED_UPDATE = "update"
SNMP_CRED_DELETE = "delete"
SNMP_DEVICE_CREATE = "create"
SNMP_DEVICE_UPDATE = "update"
SNMP_DEVICE_DELETE = "delete"
ALERT_ACK = "update"
ALERT_RESOLVE = "update"
USER_LOGIN = "login"
USER_LOGIN_FAILED = "login"
USER_LOGOUT = "logout"
USER_CREATE = "create"
USER_UPDATE_STATUS = "update"
USER_UNLOCK = "update"
USER_RESET_PASSWORD = "update"
USER_DELETE = "delete"
USER_CHANGE_PASSWORD = "update"
class AuditResource(str, enum.Enum):
"""资源类型枚举"""
USER = "user"
NETWORK = "network"
IP_ADDRESS = "ip_address"
SNMP_CREDENTIAL = "snmp_credential"
SNMP_DEVICE = "snmp_device"
ALERT = "alert"
WHITELIST_MAC = "whitelist_mac"
SCAN_TASK = "scan_task"
SYSTEM = "system"
class AuditLog(Base):
"""审计日志:记录用户的关键操作"""
"""审计日志"""
__tablename__ = "audit_logs"
id = Column(Integer, primary_key=True, index=True)
# 操作人nullable:登录失败、token 失效等场景可能没有 user_id)
user_id = Column(Integer, ForeignKey("users.id", ondelete="SET NULL"), nullable=True, index=True)
username = Column(String(50), index=True, comment="冗余存储用户名,便于 user 被删除后仍能查看")
# 操作内容
action = Column(String(50), nullable=False, index=True, comment="操作类型,如:network.create")
resource_type = Column(String(50), nullable=True, index=True, comment="资源类型:network / ip / snmp / user / scan")
resource_id = Column(String(50), nullable=True, comment="资源 ID(字符串以便兼容多种类型)")
resource_name = Column(String(200), nullable=True, comment="资源名称/标识,便于阅读")
# 操作人信息
user_id = Column(Integer, index=True, nullable=True)
username = Column(String(50), index=True)
real_name = Column(String(50))
user_ip = Column(String(50))
user_agent = Column(String(500))
# 操作信息
action = Column(Enum(AuditAction), nullable=False, index=True)
resource = Column(Enum(AuditResource), nullable=False, index=True)
resource_id = Column(String(100), nullable=True) # 操作的资源ID
# 详细信息
description = Column(String(500)) # 操作描述
old_value = Column(Text) # 修改前的值 (JSON)
new_value = Column(Text) # 修改后的值 (JSON)
changed_fields = Column(String(500)) # 变更的字段列表
# 请求信息
method = Column(String(10), comment="HTTP 方法")
path = Column(String(500), comment="请求路径")
ip_address = Column(String(50), comment="客户端 IP")
user_agent = Column(String(500), comment="User-Agent")
# 状态
status = Column(String(20), default="success", index=True, comment="success / failed")
detail = Column(Text, comment="变更详情 / 错误信息 / JSON diff")
created_at = Column(DateTime(timezone=True), server_default=func.now(), nullable=False, index=True)
# 复合索引:按时间+用户查、按时间+资源查更高效
Index("idx_audit_logs_user_created", AuditLog.user_id, AuditLog.created_at.desc())
Index("idx_audit_logs_resource", AuditLog.resource_type, AuditLog.resource_id, AuditLog.created_at.desc())
request_method = Column(String(10))
request_path = Column(String(200))
# 结果
success = Column(Integer, default=1) # 1=成功,0=失败
error_message = Column(Text)
# 时间戳
created_at = Column(DateTime(timezone=True), server_default=func.now(), index=True)
+325 -110
View File
@@ -1,13 +1,59 @@
from typing import Optional, Dict, Any, List, Tuple
from typing import Optional, List, Any
from sqlalchemy.orm import Session
from sqlalchemy import desc
from datetime import datetime
from datetime import datetime, timedelta
import json
from app.models.audit import AuditLog
from app.models.audit import AuditLog, AuditAction, AuditResource
from app.models.auth import User
# ===== 兼容层:老 API(字符串 action/resource)映射到新枚举 =====
# 字符串 action -> 枚举 + 资源类型;用于兼容各 endpoint 里硬编码的 "network.create" 等
_ACTION_MAP = {
# network
"network.create": (AuditAction.CREATE, AuditResource.NETWORK),
"network.update": (AuditAction.UPDATE, AuditResource.NETWORK),
"network.delete": (AuditAction.DELETE, AuditResource.NETWORK),
# ip
"ip.update": (AuditAction.UPDATE, AuditResource.IP_ADDRESS),
"ip.reserve": (AuditAction.UPDATE, AuditResource.IP_ADDRESS),
"ip.unreserve": (AuditAction.UPDATE, AuditResource.IP_ADDRESS),
# scan
"scan.trigger": (AuditAction.SCAN, AuditResource.SCAN_TASK),
# snmp
"snmp.credential.create": (AuditAction.CREATE, AuditResource.SNMP_CREDENTIAL),
"snmp.credential.update": (AuditAction.UPDATE, AuditResource.SNMP_CREDENTIAL),
"snmp.credential.delete": (AuditAction.DELETE, AuditResource.SNMP_CREDENTIAL),
"snmp.device.create": (AuditAction.CREATE, AuditResource.SNMP_DEVICE),
"snmp.device.update": (AuditAction.UPDATE, AuditResource.SNMP_DEVICE),
"snmp.device.delete": (AuditAction.DELETE, AuditResource.SNMP_DEVICE),
# alert
"alert.acknowledge": (AuditAction.UPDATE, AuditResource.ALERT),
"alert.resolve": (AuditAction.UPDATE, AuditResource.ALERT),
# user
"user.login": (AuditAction.LOGIN, AuditResource.USER),
"user.login.failed": (AuditAction.LOGIN, AuditResource.USER),
"user.logout": (AuditAction.LOGOUT, AuditResource.USER),
"user.create": (AuditAction.CREATE, AuditResource.USER),
"user.update_status": (AuditAction.UPDATE, AuditResource.USER),
"user.unlock": (AuditAction.UPDATE, AuditResource.USER),
"user.reset_password": (AuditAction.UPDATE, AuditResource.USER),
"user.delete": (AuditAction.DELETE, AuditResource.USER),
"user.change_password": (AuditAction.UPDATE, AuditResource.USER),
}
_RESOURCE_MAP = {
"network": AuditResource.NETWORK,
"ip": AuditResource.IP_ADDRESS,
"snmp_credential": AuditResource.SNMP_CREDENTIAL,
"snmp_device": AuditResource.SNMP_DEVICE,
"user": AuditResource.USER,
"alert": AuditResource.ALERT,
"scan": AuditResource.SCAN_TASK,
}
class AuditService:
"""审计日志服务"""
@@ -16,9 +62,9 @@ class AuditService:
db: Session,
*,
action: str,
user: Optional[User] = None,
user=None,
resource_type: Optional[str] = None,
resource_id: Optional[str] = None,
resource_id: Optional[Any] = None,
resource_name: Optional[str] = None,
method: Optional[str] = None,
path: Optional[str] = None,
@@ -26,130 +72,299 @@ class AuditService:
user_agent: Optional[str] = None,
status: str = "success",
detail: Optional[Any] = None,
) -> AuditLog:
) -> Optional["AuditLog"]:
"""
写入一条审计日志。任意字段缺失都安全降级(不抛异常),
避免审计日志写入失败影响主业务流程
兼容层:将老版 API(字符串 action/resource)映射到新版 log()。
所有 endpoint 都通过本方法写审计,避免直接调用新签名造成 500
任意环节失败都安全降级(不抛异常),审计写入失败不影响主业务。
"""
try:
log = AuditLog(
user_id=user.id if user else None,
username=user.username if user else None,
action=action,
resource_type=resource_type,
resource_id=str(resource_id) if resource_id is not None else None,
resource_name=resource_name,
method=method,
path=path,
ip_address=ip_address,
user_agent=(user_agent or "")[:500],
status=status,
detail=_serialize_detail(detail),
# action 字符串 -> (枚举, 资源枚举) ;未知 action 用 OTHER + SYSTEM
enum_action, default_resource = _ACTION_MAP.get(
action, (AuditAction.OTHER, AuditResource.SYSTEM)
)
resource_enum = _RESOURCE_MAP.get(resource_type, default_resource)
# 详情拆分:dict 拆出 old/new;其余当 description
old_value = None
new_value = None
description = None
error_message = None
success = status != "failed"
if isinstance(detail, dict):
old_value = detail.get("before")
new_value = detail.get("after")
# 把 action 原始字符串 + 资源名放到 description 里,保留细节
desc_parts = [action]
if resource_name:
desc_parts.append(resource_name)
if status == "failed":
error_message = detail.get("error") or detail.get("message") or json.dumps(detail, ensure_ascii=False, default=str)
description = " | ".join(desc_parts)
else:
description = f"{action} | {resource_name or ''}"
if status == "failed" and detail:
error_message = str(detail)
# user 不能 None 传给 log()fallback 到 User(id=None, username='anonymous')
# 但 log() 会用 user.id / user.username / user.real_name,如果都是 None 就 OK
if user is None:
anon = User(id=None, username="anonymous", real_name=None)
else:
anon = user
return AuditService.log(
db=db,
user=anon,
action=enum_action,
resource=resource_enum,
resource_id=str(resource_id) if resource_id is not None else None,
description=description[:500] if description else None,
old_value=old_value,
new_value=new_value,
user_ip=ip_address,
user_agent=(user_agent or "")[:500],
request_method=method,
request_path=path,
success=success,
error_message=error_message,
)
db.add(log)
db.commit()
return log
except Exception as e:
# 写审计日志失败不能影响主业务,仅回滚
db.rollback()
# 用 print 而非 logger,避免循环依赖(logger 可能未初始化)
print(f"[audit] failed to write log action={action}: {e}")
print(f"[audit] record() failed action={action}: {e}")
return None
@staticmethod
def get_logs(
db: Session,
*,
user_id: Optional[int] = None,
username: Optional[str] = None,
action: Optional[str] = None,
resource_type: Optional[str] = None,
status: Optional[str] = None,
start_time: Optional[datetime] = None,
end_time: Optional[datetime] = None,
skip: int = 0,
limit: int = 50,
) -> Tuple[int, List[AuditLog]]:
def log(db: Session,
user: User,
action: AuditAction,
resource: AuditResource,
resource_id: Optional[str] = None,
description: Optional[str] = None,
old_value: Optional[dict] = None,
new_value: Optional[dict] = None,
changed_fields: Optional[List[str]] = None,
user_ip: Optional[str] = None,
user_agent: Optional[str] = None,
request_method: Optional[str] = None,
request_path: Optional[str] = None,
success: bool = True,
error_message: Optional[str] = None) -> AuditLog:
"""
记录审计日志
Args:
db: 数据库会话
user: 操作用户
action: 操作类型
resource: 资源类型
resource_id: 资源ID
description: 操作描述
old_value: 旧值
new_value: 新值
changed_fields: 变更的字段列表
user_ip: 用户IP
user_agent: 用户代理
request_method: 请求方法
request_path: 请求路径
success: 操作是否成功
error_message: 错误信息
Returns:
AuditLog 实例
"""
audit_log = AuditLog(
user_id=user.id if user else None,
username=user.username if user else None,
real_name=user.real_name if user else None,
action=action,
resource=resource,
resource_id=str(resource_id) if resource_id else None,
description=description,
old_value=json.dumps(old_value, ensure_ascii=False, default=str) if old_value else None,
new_value=json.dumps(new_value, ensure_ascii=False, default=str) if new_value else None,
changed_fields=','.join(changed_fields) if changed_fields else None,
user_ip=user_ip,
user_agent=user_agent,
request_method=request_method,
request_path=request_path,
success=1 if success else 0,
error_message=error_message
)
db.add(audit_log)
db.commit()
db.refresh(audit_log)
return audit_log
@staticmethod
def get_logs(db: Session,
user_id: Optional[int] = None,
action: Optional[AuditAction] = None,
resource: Optional[AuditResource] = None,
success: Optional[bool] = None,
start_time: Optional[datetime] = None,
end_time: Optional[datetime] = None,
skip: int = 0,
limit: int = 100) -> tuple[int, List[AuditLog]]:
"""
查询审计日志
Args:
db: 数据库会话
user_id: 按用户ID过滤
action: 按操作类型过滤
resource: 按资源类型过滤
success: 按操作结果过滤
start_time: 开始时间
end_time: 结束时间
skip: 跳过记录数
limit: 返回记录数
Returns:
(总数, 记录列表)
"""
query = db.query(AuditLog)
if user_id is not None:
if user_id:
query = query.filter(AuditLog.user_id == user_id)
if username:
query = query.filter(AuditLog.username.like(f"%{username}%"))
if action:
query = query.filter(AuditLog.action.like(f"%{action}%"))
if resource_type:
query = query.filter(AuditLog.resource_type == resource_type)
if status:
query = query.filter(AuditLog.status == status)
query = query.filter(AuditLog.action == action)
if resource:
query = query.filter(AuditLog.resource == resource)
if success is not None:
query = query.filter(AuditLog.success == (1 if success else 0))
if start_time:
query = query.filter(AuditLog.created_at >= start_time)
if end_time:
query = query.filter(AuditLog.created_at <= end_time)
total = query.count()
items = query.order_by(desc(AuditLog.created_at)).offset(skip).limit(limit).all()
items = query.order_by(AuditLog.created_at.desc()).offset(skip).limit(limit).all()
return total, items
@staticmethod
def get_stats(db: Session) -> Dict[str, Any]:
"""首页/汇总用:最近 24h 操作数 + 按 action 分组"""
from datetime import timedelta
cutoff = datetime.utcnow() - timedelta(hours=24)
recent = db.query(AuditLog).filter(AuditLog.created_at >= cutoff).count()
by_action = db.query(AuditLog.action, db.query(AuditLog).filter(AuditLog.created_at >= cutoff).subquery()) # 占位避免循环
# 简化:用 group by
from sqlalchemy import func
rows = db.query(AuditLog.action, func.count(AuditLog.id)).filter(
AuditLog.created_at >= cutoff
).group_by(AuditLog.action).all()
def get_user_actions(db: Session, user_id: int, limit: int = 50) -> List[AuditLog]:
"""获取指定用户的操作日志"""
return db.query(AuditLog)\
.filter(AuditLog.user_id == user_id)\
.order_by(AuditLog.created_at.desc())\
.limit(limit)\
.all()
@staticmethod
def get_resource_logs(db: Session, resource: AuditResource, resource_id: str, limit: int = 50) -> List[AuditLog]:
"""获取指定资源的操作历史"""
return db.query(AuditLog)\
.filter(AuditLog.resource == resource, AuditLog.resource_id == str(resource_id))\
.order_by(AuditLog.created_at.desc())\
.limit(limit)\
.all()
@staticmethod
def get_statistics(db: Session, days: int = 30) -> dict:
"""
获取审计统计信息
Args:
db: 数据库会话
days: 统计天数
Returns:
统计信息字典
"""
start_date = datetime.utcnow() - timedelta(days=days)
# 总操作数
total_ops = db.query(AuditLog).filter(AuditLog.created_at >= start_date).count()
# 成功/失败统计
success_ops = db.query(AuditLog).filter(
AuditLog.created_at >= start_date,
AuditLog.success == 1
).count()
failed_ops = total_ops - success_ops
# 按操作类型统计
action_stats = {}
for action in AuditAction:
count = db.query(AuditLog).filter(
AuditLog.created_at >= start_date,
AuditLog.action == action
).count()
if count > 0:
action_stats[action.value] = count
# 按资源类型统计
resource_stats = {}
for resource in AuditResource:
count = db.query(AuditLog).filter(
AuditLog.created_at >= start_date,
AuditLog.resource == resource
).count()
if count > 0:
resource_stats[resource.value] = count
# 活跃用户数
active_users = db.query(AuditLog.user_id).filter(
AuditLog.created_at >= start_date
).distinct().count()
return {
"recent_24h": recent,
"by_action_24h": {action: count for action, count in rows},
"period_days": days,
"total_operations": total_ops,
"successful_operations": success_ops,
"failed_operations": failed_ops,
"success_rate": round(success_ops / total_ops * 100, 2) if total_ops > 0 else 0,
"by_action": action_stats,
"by_resource": resource_stats,
"active_users_count": active_users
}
@staticmethod
def clean_old_logs(db: Session, days: int = 90) -> int:
"""
清理指定天数之前的日志
Args:
db: 数据库会话
days: 保留天数
Returns:
删除的记录数
"""
cutoff_date = datetime.utcnow() - timedelta(days=days)
deleted = db.query(AuditLog).filter(AuditLog.created_at < cutoff_date).delete()
db.commit()
return deleted
def _serialize_detail(detail: Any) -> Optional[str]:
"""把 dict/list 等结构化数据序列化成字符串,便于存储和展示"""
if detail is None:
return None
if isinstance(detail, str):
return detail[:4000]
try:
return json.dumps(detail, ensure_ascii=False, default=str)[:4000]
except (TypeError, ValueError):
return str(detail)[:4000]
# 常用 action 字符串集中管理(避免散落各处的魔法字符串)
class AuditAction:
# 网络
NETWORK_CREATE = "network.create"
NETWORK_UPDATE = "network.update"
NETWORK_DELETE = "network.delete"
# IP
IP_UPDATE = "ip.update"
IP_RESERVE = "ip.reserve"
IP_UNRESERVE = "ip.unreserve"
# 扫描
SCAN_TRIGGER = "scan.trigger"
# SNMP
SNMP_CRED_CREATE = "snmp.credential.create"
SNMP_CRED_UPDATE = "snmp.credential.update"
SNMP_CRED_DELETE = "snmp.credential.delete"
SNMP_DEVICE_CREATE = "snmp.device.create"
SNMP_DEVICE_UPDATE = "snmp.device.update"
SNMP_DEVICE_DELETE = "snmp.device.delete"
# 告警
ALERT_ACK = "alert.acknowledge"
ALERT_RESOLVE = "alert.resolve"
# 用户管理
USER_LOGIN = "user.login"
USER_LOGIN_FAILED = "user.login.failed"
USER_LOGOUT = "user.logout"
USER_CREATE = "user.create"
USER_UPDATE_STATUS = "user.update_status"
USER_UNLOCK = "user.unlock"
USER_RESET_PASSWORD = "user.reset_password"
USER_DELETE = "user.delete"
USER_CHANGE_PASSWORD = "user.change_password"
class AuditLogger:
"""审计日志装饰器/上下文管理器"""
def __init__(self, db: Session, user: User, resource: AuditResource, resource_id: Optional[str] = None):
self.db = db
self.user = user
self.resource = resource
self.resource_id = resource_id
self.old_value = None
def set_old_value(self, value: dict):
"""设置旧值(用于更新操作)"""
self.old_value = value
def log(self, action: AuditAction, description: str, new_value: Optional[dict] = None, changed_fields: Optional[List[str]] = None):
"""记录日志"""
AuditService.log(
db=self.db,
user=self.user,
action=action,
resource=self.resource,
resource_id=self.resource_id,
description=description,
old_value=self.old_value,
new_value=new_value,
changed_fields=changed_fields
)
+239
View File
@@ -0,0 +1,239 @@
from pydantic_settings import BaseSettings
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
import smtplib
from typing import Optional, List
from datetime import datetime
import logging
logger = logging.getLogger(__name__)
class EmailSettings(BaseSettings):
"""邮件配置"""
SMTP_HOST: str = "localhost"
SMTP_PORT: int = 25
SMTP_USER: Optional[str] = None
SMTP_PASSWORD: Optional[str] = None
SMTP_USE_TLS: bool = False
SMTP_USE_SSL: bool = False
SMTP_FROM: str = "ipam@localhost"
class Config:
env_prefix = "EMAIL_"
class EmailService:
"""邮件服务"""
def __init__(self, settings: EmailSettings):
self.settings = settings
def send_email(self, to: str, subject: str, html_content: str) -> bool:
"""
发送HTML邮件
Args:
to: 收件人邮箱
subject: 邮件主题
html_content: HTML内容
Returns:
是否发送成功
"""
try:
msg = MIMEMultipart('alternative')
msg['From'] = self.settings.SMTP_FROM
msg['To'] = to
msg['Subject'] = subject
html_part = MIMEText(html_content, 'html', 'utf-8')
msg.attach(html_part)
# 连接SMTP服务器
if self.settings.SMTP_USE_SSL:
server = smtplib.SMTP_SSL(
host=self.settings.SMTP_HOST,
port=self.settings.SMTP_PORT,
timeout=10
)
else:
server = smtplib.SMTP(
host=self.settings.SMTP_HOST,
port=self.settings.SMTP_PORT,
timeout=10
)
# 如果需要TLS加密
if self.settings.SMTP_USE_TLS and not self.settings.SMTP_USE_SSL:
server.starttls()
# 如果需要认证
if self.settings.SMTP_USER and self.settings.SMTP_PASSWORD:
server.login(self.settings.SMTP_USER, self.settings.SMTP_PASSWORD)
# 发送邮件
server.send_message(msg)
server.quit()
logger.info(f"邮件发送成功: {to} - {subject}")
return True
except Exception as e:
logger.error(f"邮件发送失败: {str(e)}", exc_info=True)
return False
def send_alert_notification(self, to: str, alert_data: dict) -> bool:
"""发送告警通知邮件"""
severity_colors = {
'critical': '#dc2626',
'error': '#ef4444',
'warning': '#f59e0b',
'info': '#3b82f6'
}
severity_texts = {
'critical': '严重',
'error': '错误',
'warning': '警告',
'info': '信息'
}
alert_type_texts = {
'ip_conflict': 'IP 地址冲突',
'unauthorized_access': '未授权设备接入',
'subnet_full': '网段容量不足',
'device_offline': '设备离线',
'new_device_detected': '新设备发现',
'mac_changed': 'MAC 地址变更'
}
severity = alert_data.get('severity', 'warning')
color = severity_colors.get(severity, '#6b7280')
severity_text = severity_texts.get(severity, severity)
alert_type_text = alert_type_texts.get(alert_data.get('alert_type'), alert_data.get('alert_type'))
html_content = f"""
<html>
<head>
<meta charset="UTF-8">
<style>
body {{ font-family: Arial, sans-serif; line-height: 1.6; color: #333; }}
.header {{ background-color: {color}; color: white; padding: 20px; border-radius: 5px; }}
.content {{ padding: 20px; background-color: #f9fafb; border-radius: 5px; margin-top: 10px; }}
.alert-item {{ margin-bottom: 15px; }}
.label {{ font-weight: bold; color: #6b7280; display: inline-block; width: 100px; }}
.footer {{ margin-top: 20px; padding-top: 20px; border-top: 1px solid #e5e7eb; font-size: 12px; color: #9ca3af; }}
</style>
</head>
<body>
<div class="header">
<h2>⚠️ IPAM 告警通知</h2>
<p>告警级别: <strong>{severity_text}</strong></p>
</div>
<div class="content">
<div class="alert-item">
<span class="label">告警类型:</span> {alert_type_text}
</div>
<div class="alert-item">
<span class="label">告警标题:</span> {alert_data.get('title', '-')}
</div>
<div class="alert-item">
<span class="label">详细信息:</span> {alert_data.get('message', '-')}
</div>
<div class="alert-item">
<span class="label">相关IP:</span> {alert_data.get('ip_address_str', '-')}
</div>
<div class="alert-item">
<span class="label">相关MAC:</span> {alert_data.get('mac_address', '-')}
</div>
<div class="alert-item">
<span class="label">告警时间:</span> {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
</div>
</div>
<div class="footer">
<p>此邮件由 IPAM 地址管理系统自动发送,请勿直接回复。</p>
<p>如需处理该告警,请登录 IPAM 管理控制台: <a href="#">点击访问</a></p>
</div>
</body>
</html>
"""
subject = f"{severity_text}】IPAM 告警 - {alert_data.get('title', '新告警')}"
return self.send_email(to, subject, html_content)
def send_scan_report(self, to: str, report_data: dict) -> bool:
"""发送扫描报告邮件"""
html_content = f"""
<html>
<head>
<meta charset="UTF-8">
<style>
body {{ font-family: Arial, sans-serif; line-height: 1.6; color: #333; }}
.header {{ background-color: #3b82f6; color: white; padding: 20px; border-radius: 5px; }}
.content {{ padding: 20px; background-color: #f9fafb; border-radius: 5px; margin-top: 10px; }}
.stat-item {{ display: inline-block; width: 22%; text-align: center; padding: 15px; background: white; border-radius: 5px; margin: 1%; }}
.stat-number {{ font-size: 28px; font-weight: bold; color: #3b82f6; }}
.stat-label {{ color: #6b7280; font-size: 14px; }}
.success {{ color: #10b981; }}
.warning {{ color: #f59e0b; }}
.footer {{ margin-top: 20px; padding-top: 20px; border-top: 1px solid #e5e7eb; font-size: 12px; color: #9ca3af; }}
</style>
</head>
<body>
<div class="header">
<h2>📊 IPAM 网段扫描报告</h2>
</div>
<div class="content">
<h3>扫描结果概览</h3>
<div class="stat-item">
<div class="stat-number">{report_data.get('total_ips', 0)}</div>
<div class="stat-label">扫描IP总数</div>
</div>
<div class="stat-item">
<div class="stat-number success">{report_data.get('online_ips', 0)}</div>
<div class="stat-label">在线设备</div>
</div>
<div class="stat-item">
<div class="stat-number">{report_data.get('with_mac', 0)}</div>
<div class="stat-label">发现MAC地址</div>
</div>
<div class="stat-item">
<div class="stat-number warning">{report_data.get('new_devices', 0)}</div>
<div class="stat-label">新发现设备</div>
</div>
<div style="clear: both; margin-top: 20px;">
<p><strong>扫描网段:</strong> {report_data.get('network_cidr', '-')}</p>
<p><strong>扫描时间:</strong> {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</p>
</div>
</div>
<div class="footer">
<p>此邮件由 IPAM 地址管理系统自动发送,请勿直接回复。</p>
</div>
</body>
</html>
"""
subject = f"IPAM 扫描报告 - {report_data.get('network_cidr', '')}"
return self.send_email(to, subject, html_content)
# 全局邮件服务实例
_email_service = None
def get_email_service() -> EmailService:
"""获取邮件服务实例"""
global _email_service
if _email_service is None:
settings = EmailSettings()
_email_service = EmailService(settings)
return _email_service
def init_email_service(settings: EmailSettings):
"""初始化邮件服务"""
global _email_service
_email_service = EmailService(settings)
+159 -72
View File
@@ -6,96 +6,102 @@ import subprocess
import socket
import ipaddress
import re
import logging
try:
from mac_vendor_lookup import MacLookup, VendorNotFoundError
MAC_LOOKUP_AVAILABLE = True
except ImportError:
MAC_LOOKUP_AVAILABLE = False
from app.models.network import ScanTask, Network, IPAddress
from app.models.network import TaskStatus, TaskType, IPStatus
from app.core.config import settings
logger = __import__('logging').getLogger(__name__)
logger = logging.getLogger(__name__)
class EnhancedScanService:
"""增强扫描服务 - 包含ARP、DNS解析、MAC厂商识别"""
"""增强扫描服务 - 包含ARP、DNS解析、MAC厂商识别、NetBIOS主机名发现"""
# MAC OUI 厂商映射(常用厂商
OUI_MAP = {
'00:1C:14': 'VMware',
'00:0C:29': 'VMware',
'00:50:56': 'VMware',
'00:15:5D': 'Microsoft',
'08:00:27': 'VirtualBox',
'52:54:00': 'QEMU/KVM',
'00:1A:2A': 'Dell',
'00:21:9B': 'Dell',
'00:26:B9': 'Dell',
'00:1B:21': 'HP',
'00:23:7D': 'HP',
'00:1E:0B': 'Cisco',
'00:21:55': 'Cisco',
'00:23:04': 'Cisco',
'00:24:14': 'Cisco',
'00:1F:CA': 'Cisco',
'00:19:BB': 'Huawei',
'00:25:9E': 'Huawei',
'00:E0:FC': 'Huawei',
'AC:85:3D': 'Huawei',
'28:6E:D4': 'Huawei',
'00:16:6F': 'H3C',
'00:23:89': 'H3C',
'00:0F:E2': 'Intel',
'00:1B:77': 'Intel',
'00:21:6A': 'Intel',
'1C:6F:65': 'Intel',
'00:19:D2': 'Realtek',
'00:24:1D': 'Realtek',
'00:E0:4C': 'Realtek',
'F4:6D:04': 'Apple',
'E0:F8:47': 'Apple',
'C8:BC:C8': 'Apple',
'04:15:52': 'Apple',
'28:CF:E9': 'Apple',
'04:7D:7B': 'Xiaomi',
'18:59:36': 'Xiaomi',
'34:CE:00': 'Xiaomi',
'64:09:80': 'TP-Link',
'E8:94:F6': 'TP-Link',
'50:FA:84': 'TP-Link',
'00:24:01': 'TP-Link',
'30:FC:68': 'NETGEAR',
'00:09:5B': 'NETGEAR',
'00:FF:FF': 'Broadcast',
'FF:FF:FF': 'Broadcast',
}
# 初始化 MAC 厂商查询(单例,共享 OUI 数据库
_mac_lookup = None
@classmethod
def _get_mac_lookup(cls):
"""获取或初始化 MacLookup 实例"""
if cls._mac_lookup is None and MAC_LOOKUP_AVAILABLE:
cls._mac_lookup = MacLookup()
return cls._mac_lookup
@staticmethod
def get_mac_vendor(mac_address: str) -> Optional[str]:
"""根据MAC地址OUI识别厂商"""
"""根据 MAC 地址 OUI 识别厂商"""
if not mac_address:
return None
# 标准化MAC格式为 AA:BB:CC:DD:EE:FF
# 标准化 MAC 格式为 AA:BB:CC:DD:EE:FF
mac = mac_address.upper().replace('-', ':')
if len(mac) < 8:
return None
# 提取前3字节 OUI
# 使用 mac_vendor_lookup 库(IEEE OUI 官方数据库)
if MAC_LOOKUP_AVAILABLE:
lookup = EnhancedScanService._get_mac_lookup()
if lookup is not None:
try:
# 提取 OUI(前 3 字节)
oui = mac[:8] # "AA:BB:CC"
vendor = lookup.lookup(oui)
if vendor:
return vendor
except (VendorNotFoundError, AttributeError, TypeError):
pass
except Exception:
pass
# 回退:硬编码常用厂商映射
return EnhancedScanService._legacy_oui_lookup(mac)
# 硬编码回退映射(mac_vendor_lookup 不可用时使用)
_LEGACY_OUI_MAP = {
'00:1C:14': 'VMware', '00:0C:29': 'VMware', '00:50:56': 'VMware',
'00:15:5D': 'Microsoft', '08:00:27': 'VirtualBox', '52:54:00': 'QEMU/KVM',
'00:1A:2A': 'Dell', '00:21:9B': 'Dell', '00:26:B9': 'Dell',
'00:1B:21': 'HP', '00:23:7D': 'HP',
'00:1E:0B': 'Cisco', '00:21:55': 'Cisco', '00:23:04': 'Cisco',
'00:24:14': 'Cisco', '00:1F:CA': 'Cisco',
'00:19:BB': 'Huawei', '00:25:9E': 'Huawei', '00:E0:FC': 'Huawei',
'AC:85:3D': 'Huawei', '28:6E:D4': 'Huawei',
'00:16:6F': 'H3C', '00:23:89': 'H3C',
'00:0F:E2': 'Intel', '00:1B:77': 'Intel', '00:21:6A': 'Intel',
'1C:6F:65': 'Intel',
'00:19:D2': 'Realtek', '00:24:1D': 'Realtek', '00:E0:4C': 'Realtek',
'F4:6D:04': 'Apple', 'E0:F8:47': 'Apple', 'C8:BC:C8': 'Apple',
'04:15:52': 'Apple', '28:CF:E9': 'Apple',
'04:7D:7B': 'Xiaomi', '18:59:36': 'Xiaomi', '34:CE:00': 'Xiaomi',
'64:09:80': 'TP-Link', 'E8:94:F6': 'TP-Link', '50:FA:84': 'TP-Link',
'00:24:01': 'TP-Link',
'30:FC:68': 'NETGEAR', '00:09:5B': 'NETGEAR',
'00:FF:FF': 'Broadcast', 'FF:FF:FF': 'Broadcast',
}
@staticmethod
def _legacy_oui_lookup(mac: str) -> Optional[str]:
"""硬编码 OUI 回退查询"""
oui = mac[:8]
# 精确匹配
if oui in EnhancedScanService.OUI_MAP:
return EnhancedScanService.OUI_MAP[oui]
# 尝试模糊匹配(前2字节)
if oui in EnhancedScanService._LEGACY_OUI_MAP:
return EnhancedScanService._LEGACY_OUI_MAP[oui]
# 模糊匹配(前 2 字节)
oui_short = mac[:5]
for key, vendor in EnhancedScanService.OUI_MAP.items():
for key, vendor in EnhancedScanService._LEGACY_OUI_MAP.items():
if key.startswith(oui_short):
return vendor
return None
@staticmethod
def reverse_dns_lookup(ip_address: str, timeout: int = 2) -> Optional[str]:
"""反向DNS解析获取主机名"""
"""反向 DNS 解析获取主机名"""
try:
socket.setdefaulttimeout(timeout)
hostname, _, _ = socket.gethostbyaddr(ip_address)
@@ -103,6 +109,64 @@ class EnhancedScanService:
except (socket.herror, socket.timeout, socket.gaierror):
return None
@staticmethod
def netbios_lookup(ip_address: str, timeout: int = 3) -> Optional[str]:
"""
通过 NetBIOS 广播协议获取主机名(Windows / SMB 设备)
使用 nmblookup -A <ip> 查询
"""
try:
result = subprocess.run(
['nmblookup', '-A', ip_address],
capture_output=True, text=True, timeout=timeout
)
if result.returncode == 0:
# nmblookup 输出格式示例:
# querying host 192.168.1.100 on 192.168.1.255
#
# WORKGROUP<00> - <ACTIVE> B
# WORKGROUP<00> - <GROUP> B
# WORKGROUP<1D> - <GROUP> B
# SERVER01<20> - <ACTIVE> B
# 主机名通常是 <20> 服务的第一列
for line in result.stdout.strip().split('\n'):
line = line.strip()
if not line or line.startswith('querying'):
continue
# 匹配 "HOSTNAME<20> - <ACTIVE> B" 格式
match = re.match(
r'^(\S+?)<20>\s+-\s+<ACTIVE>',
line,
re.IGNORECASE
)
if match:
return match.group(1).upper()
except Exception:
pass
return None
@staticmethod
def snmp_sysname_lookup(device_id: int, db: Session) -> Optional[str]:
"""
通过 SNMP sysName 获取主机名(网络设备 / 服务器)
仅在扫描时已配置 SNMP 凭据的设备上调用
"""
from app.models.snmp import NetworkDevice
from app.services.snmp_service import SNMPService
try:
device = db.query(NetworkDevice).filter(
NetworkDevice.id == device_id
).first()
if device and device.snmp_credential:
success, info = SNMPService.test_connection(
device, device.snmp_credential
)
if success:
return info.get('sys_name')
except Exception:
pass
return None
@staticmethod
def arp_scan_single(ip_address: str, timeout: int = 2) -> Optional[str]:
"""
@@ -116,7 +180,6 @@ class EnhancedScanService:
capture_output=True, text=True, timeout=timeout + 2
)
if result.returncode == 0:
# arp-scan 输出格式: IP\tMAC\tVendor
for line in result.stdout.strip().split('\n'):
line = line.strip()
if not line:
@@ -155,7 +218,6 @@ class EnhancedScanService:
timeout=2
)
if result.returncode == 0:
# 匹配 MAC 地址格式
mac_pattern = r'([0-9A-Fa-f]{2}[:-]){5}([0-9A-Fa-f]{2})'
match = re.search(mac_pattern, result.stdout)
if match:
@@ -170,9 +232,20 @@ class EnhancedScanService:
enable_ping: bool = True,
enable_arp: bool = True,
enable_dns: bool = True,
enable_netbios: bool = True,
timeout: int = 2) -> Dict[str, Any]:
"""
综合扫描单个IP
厂商识别:
- 主要: mac_vendor_lookup 库(IEEE OUI 官方数据库,~17万条)
- 回退: 硬编码 OUI_MAP
主机名发现(多源,按优先级):
1. 反向 DNS 解析(PTR
2. NetBIOS 广播(Windows / SMB 设备)
3. SNMP sysName(网络设备 / 服务器,需配置 SNMP)
返回: {
'ip_address': str,
'status': 'online'|'offline',
@@ -203,14 +276,21 @@ class EnhancedScanService:
mac = EnhancedScanService.arp_scan_single(ip_address, timeout)
if mac:
result['mac_address'] = mac
# 厂商识别:优先 mac_vendor_lookup 库,失败后回退硬编码映射
result['vendor'] = EnhancedScanService.get_mac_vendor(mac)
# 如果获取到MAC,说明设备实际上在线
result['status'] = 'online'
result['success'] = True
# 3. 反向 DNS 解析
if enable_dns and result.get('success'):
hostname = EnhancedScanService.reverse_dns_lookup(ip_address, timeout)
# 3. 主机名发现(多源,有任一在线即执行)
if result.get('success'):
hostname = None
# 3a. 反向 DNS
if enable_dns:
hostname = EnhancedScanService.reverse_dns_lookup(ip_address, timeout)
# 3b. NetBIOSWindows / SMB
if not hostname and enable_netbios:
hostname = EnhancedScanService.netbios_lookup(ip_address, timeout=3)
if hostname:
result['hostname'] = hostname
@@ -249,6 +329,7 @@ class EnhancedScanService:
enable_ping: bool = True,
enable_arp: bool = True,
enable_dns: bool = True,
enable_netbios: bool = True,
timeout: int = 2,
max_concurrent: int = 50) -> List[Dict[str, Any]]:
"""
@@ -260,7 +341,6 @@ class EnhancedScanService:
if not ips:
return []
# 并发执行;并发数随网段大小自适应
worker_count = min(max_concurrent, len(ips))
results: List[Dict[str, Any]] = []
with ThreadPoolExecutor(max_workers=worker_count) as executor:
@@ -271,6 +351,7 @@ class EnhancedScanService:
enable_ping=enable_ping,
enable_arp=enable_arp,
enable_dns=enable_dns,
enable_netbios=enable_netbios,
timeout=timeout
): ip
for ip in ips
@@ -303,12 +384,10 @@ class EnhancedScanService:
# 如果IP不存在,尝试自动创建
if not db_ip:
# 从ip_address推断network_id
network = db.query(Network).filter(
Network.cidr.op('@>')((ip_address + '/32').encode('utf-8'))
).first()
if not network:
# 尝试精确匹配 network.cidr
for net in db.query(Network).all():
try:
if ip_address in ipaddress.ip_network(net.cidr, strict=False):
@@ -342,11 +421,15 @@ class EnhancedScanService:
if hostname:
db_ip.hostname = hostname
# 更新厂商信息
# 更新厂商信息(即使没有 MAC,也尝试用现有 MAC 补全)
vendor = scan_result.get('vendor')
if vendor:
db_ip.vendor = vendor
# 如果没有拿到 vendor 但有 MAC,用现有 MAC 补全厂商
if not db_ip.vendor and db_ip.mac_address:
db_ip.vendor = EnhancedScanService.get_mac_vendor(db_ip.mac_address)
# 更新最后发现时间
if scan_result.get('success'):
db_ip.last_seen = datetime.utcnow()
@@ -360,6 +443,7 @@ class EnhancedScanService:
mac_found_count = 0
hostname_found_count = 0
created_count = 0
vendor_found_count = 0
for result in scan_results:
ip = EnhancedScanService.update_ip_from_scan_result(db, result)
@@ -370,6 +454,8 @@ class EnhancedScanService:
mac_found_count += 1
if result.get('hostname'):
hostname_found_count += 1
if result.get('vendor'):
vendor_found_count += 1
db.commit()
@@ -377,5 +463,6 @@ class EnhancedScanService:
'online_count': online_count,
'mac_found_count': mac_found_count,
'hostname_found_count': hostname_found_count,
'vendor_found_count': vendor_found_count,
'total_scanned': len(scan_results)
}
+277
View File
@@ -0,0 +1,277 @@
from sqlalchemy.orm import Session
from typing import Optional, List, Dict, Any
from io import BytesIO, StringIO
import csv
from datetime import datetime
from app.models.network import IPAddress, Network, IPStatus
from app.models.snmp import NetworkDevice, ARPEntry
from app.models.alert import Alert
class ReportExportService:
"""报表导出服务"""
# ========== CSV 导出 ==========
@staticmethod
def export_ip_addresses_csv(db: Session,
network_id: Optional[int] = None,
status: Optional[IPStatus] = None,
include_online_only: bool = False) -> tuple[str, str]:
"""
导出IP地址表为CSV
Args:
db: 数据库会话
network_id: 网段ID(可选,不指定则导出所有)
status: IP状态过滤
include_online_only: 仅导出在线IP
Returns:
(CSV内容字符串, 文件名)
"""
query = db.query(IPAddress)
if network_id:
query = query.filter(IPAddress.network_id == network_id)
if status:
query = query.filter(IPAddress.status == status)
if include_online_only:
query = query.filter(IPAddress.status == IPStatus.ONLINE)
ips = query.order_by(IPAddress.network_id, IPAddress.id).all()
output = StringIO()
writer = csv.writer(output)
# 写入表头
writer.writerow([
'ID', '网段ID', 'IP地址', 'MAC地址', '主机名', '状态',
'使用者', '业务类型', '备注', '厂商',
'最后发现时间', '创建时间'
])
# 写入数据
for ip in ips:
writer.writerow([
ip.id,
ip.network_id,
ip.ip_address,
ip.mac_address or '',
ip.hostname or '',
ip.status.value,
ip.owner or '',
ip.business_type or '',
ip.notes or '',
ip.vendor or '',
ip.last_seen.strftime('%Y-%m-%d %H:%M:%S') if ip.last_seen else '',
ip.created_at.strftime('%Y-%m-%d %H:%M:%S') if ip.created_at else ''
])
csv_content = output.getvalue()
filename = f"ip_addresses_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
return csv_content, filename
@staticmethod
def export_networks_csv(db: Session) -> tuple[str, str]:
"""导出网段汇总CSV"""
networks = db.query(Network).order_by(Network.id).all()
output = StringIO()
writer = csv.writer(output)
writer.writerow([
'ID', '网段CIDR', '名称', '描述', '分组',
'网关', 'VLAN', '总IP数', '已用IP数', '保留IP数',
'创建时间'
])
for net in networks:
writer.writerow([
net.id,
net.cidr,
net.name or '',
net.description or '',
net.group_name or '',
net.gateway or '',
net.vlan_id or '',
net.total_ips,
net.used_ips,
net.reserved_ips,
net.created_at.strftime('%Y-%m-%d %H:%M:%S') if net.created_at else ''
])
csv_content = output.getvalue()
filename = f"networks_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
return csv_content, filename
@staticmethod
def export_alerts_csv(db: Session, status: Optional[str] = None) -> tuple[str, str]:
"""导出告警CSV"""
query = db.query(Alert)
if status:
query = query.filter(Alert.status == status)
alerts = query.order_by(Alert.created_at.desc()).all()
output = StringIO()
writer = csv.writer(output)
writer.writerow([
'ID', '告警类型', '严重程度', '状态', '标题',
'详细信息', '相关IP', '相关MAC', '创建时间'
])
for alert in alerts:
writer.writerow([
alert.id,
alert.alert_type.value,
alert.severity.value,
alert.status.value,
alert.title,
alert.message or '',
alert.ip_address_str or '',
alert.mac_address or '',
alert.created_at.strftime('%Y-%m-%d %H:%M:%S') if alert.created_at else ''
])
csv_content = output.getvalue()
filename = f"alerts_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
return csv_content, filename
@staticmethod
def export_snmp_devices_csv(db: Session) -> tuple[str, str]:
"""导出SNMP设备CSV"""
devices = db.query(NetworkDevice).order_by(NetworkDevice.id).all()
output = StringIO()
writer = csv.writer(output)
writer.writerow([
'ID', '设备名称', 'IP地址', '端口', '设备类型',
'厂商', '型号', '位置', '最后轮询时间', '创建时间'
])
for dev in devices:
writer.writerow([
dev.id,
dev.name,
dev.ip_address,
dev.port,
dev.device_type.value,
dev.vendor or '',
dev.model or '',
dev.location or '',
dev.last_polled_at.strftime('%Y-%m-%d %H:%M:%S') if dev.last_polled_at else '',
dev.created_at.strftime('%Y-%m-%d %H:%M:%S') if dev.created_at else ''
])
csv_content = output.getvalue()
filename = f"snmp_devices_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
return csv_content, filename
@staticmethod
def export_arp_table_csv(db: Session, device_id: Optional[int] = None) -> tuple[str, str]:
"""导出ARP表CSV"""
query = db.query(ARPEntry)
if device_id:
query = query.filter(ARPEntry.device_id == device_id)
entries = query.order_by(ARPEntry.device_id, ARPEntry.id).all()
output = StringIO()
writer = csv.writer(output)
writer.writerow([
'ID', '设备ID', 'IP地址', 'MAC地址', '接口', '最后发现时间'
])
for entry in entries:
writer.writerow([
entry.id,
entry.device_id,
entry.ip_address,
entry.mac_address,
entry.interface or '',
entry.last_seen.strftime('%Y-%m-%d %H:%M:%S') if entry.last_seen else ''
])
csv_content = output.getvalue()
filename = f"arp_table_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
return csv_content, filename
# ========== 汇总统计报表 ==========
@staticmethod
def get_ipam_summary(db: Session) -> Dict[str, Any]:
"""获取IPAM系统汇总报表"""
# 网段统计
total_networks = db.query(Network).count()
total_ips = db.query(IPAddress).count()
online_ips = db.query(IPAddress).filter(IPAddress.status == IPStatus.ONLINE).count()
reserved_ips = db.query(IPAddress).filter(IPAddress.status == IPStatus.RESERVED).count()
# 带MAC/主机名的IP统计
with_mac = db.query(IPAddress).filter(IPAddress.mac_address.isnot(None)).count()
with_hostname = db.query(IPAddress).filter(IPAddress.hostname.isnot(None)).count()
# 网段使用率分布
networks = db.query(Network).all()
usage_distribution = {
'under_50': 0, # 50%以下
'50_70': 0, # 50%-70%
'70_90': 0, # 70%-90%
'over_90': 0 # 90%以上
}
for net in networks:
usage_rate = (net.used_ips / net.total_ips * 100) if net.total_ips > 0 else 0
if usage_rate < 50:
usage_distribution['under_50'] += 1
elif usage_rate < 70:
usage_distribution['50_70'] += 1
elif usage_rate < 90:
usage_distribution['70_90'] += 1
else:
usage_distribution['over_90'] += 1
# 告警统计
from app.models.alert import Alert, AlertStatus
active_alerts = db.query(Alert).filter(Alert.status == AlertStatus.ACTIVE).count()
resolved_alerts = db.query(Alert).filter(Alert.status == AlertStatus.RESOLVED).count()
# SNMP设备统计
snmp_devices = db.query(NetworkDevice).count()
active_devices = db.query(NetworkDevice).filter(NetworkDevice.last_polled_at.isnot(None)).count()
return {
"summary": {
"total_networks": total_networks,
"total_ip_addresses": total_ips,
"online_ips": online_ips,
"reserved_ips": reserved_ips,
"available_ips": total_ips - online_ips - reserved_ips,
"online_rate": round(online_ips / total_ips * 100, 2) if total_ips > 0 else 0
},
"discovery": {
"with_mac_address": with_mac,
"with_hostname": with_hostname,
"mac_discovery_rate": round(with_mac / total_ips * 100, 2) if total_ips > 0 else 0
},
"network_usage_distribution": usage_distribution,
"alerts": {
"active": active_alerts,
"resolved": resolved_alerts
},
"snmp": {
"total_devices": snmp_devices,
"active_devices": active_devices
},
"generated_at": datetime.now().strftime('%Y-%m-%d %H:%M:%S')
}
+4 -2
View File
@@ -221,18 +221,20 @@ class SNMPService:
arp_data[ip_address]['if_index'] = if_index
# 转换为列表
now = datetime.utcnow()
for ip, data in arp_data.items():
if 'mac_address' in data:
entry = {
'ip_address': ip,
'mac_address': data.get('mac_address'),
'interface': str(data.get('if_index', '')),
'device_id': device.id
'device_id': device.id,
'last_seen': now.isoformat(),
'discovered_at': now.isoformat()
}
arp_entries.append(entry)
# 保存到数据库
now = datetime.utcnow()
for entry_data in arp_entries:
# 查找是否已存在
existing = db.query(ARPEntry).filter(
+10 -3
View File
@@ -21,6 +21,7 @@ def register_tasks(celery_app):
enable_ping: bool = True,
enable_arp: bool = True,
enable_dns: bool = True,
enable_netbios: bool = True,
timeout: int = 2):
"""
Celery 异步任务:扫描指定网段
@@ -54,6 +55,7 @@ def register_tasks(celery_app):
enable_ping=enable_ping,
enable_arp=enable_arp,
enable_dns=enable_dns,
enable_netbios=enable_netbios,
timeout=timeout
)
@@ -120,7 +122,8 @@ def register_tasks(celery_app):
@celery_app.task(bind=True)
def full_network_scan(self, enable_ping: bool = True,
enable_arp: bool = True,
enable_dns: bool = True):
enable_dns: bool = True,
enable_netbios: bool = True):
"""
Celery 定时任务:扫描所有网段
"""
@@ -144,7 +147,8 @@ def register_tasks(celery_app):
network.cidr,
enable_ping=enable_ping,
enable_arp=enable_arp,
enable_dns=enable_dns
enable_dns=enable_dns,
enable_netbios=enable_netbios
)
stats = EnhancedScanService.bulk_update_ips_from_scan(db, results)
@@ -191,7 +195,10 @@ def register_tasks(celery_app):
results = []
for ip in ip_addresses:
result = EnhancedScanService.comprehensive_scan(ip, enable_ping=True, enable_arp=True, enable_dns=False)
result = EnhancedScanService.comprehensive_scan(
ip, enable_ping=True, enable_arp=True, enable_dns=False,
enable_netbios=True
)
EnhancedScanService.update_ip_from_scan_result(db, result)
results.append(result)
+95
View File
@@ -0,0 +1,95 @@
# IPAM 运维脚本
存放 IPAM 平台的一次性运维脚本。所有脚本都假定从 `backend/` 目录执行,且 venv 已激活。
## 回填厂商 / 主机名 — `backfill_vendor.py`
### 用途
在某些情况下(升级前扫描失败、OUI 库没装、依赖 bug 等)IP 表里 `mac_address` 有值但 `vendor`/`hostname` 是空。本脚本会:
- 找出所有 `mac_address IS NOT NULL AND vendor IS NULL` 的 IP
- 调用 `EnhancedScanService.get_mac_vendor()` 重新识别厂商(基于 IEEE OUI 数据库)
- 可选:调用 `reverse_dns_lookup()` 回填主机名
- 支持 dry-run 模式(只统计不写库)
### 用法
```bash
# 1. 确保依赖装了(首次或升级后必做)
pip install mac-vendor-lookup
# 2. 干跑:只统计能识别的 IP 数量,不写库
python scripts/backfill_vendor.py --dry-run
# 3. 实际回填
python scripts/backfill_vendor.py
# 4. 同时回填主机名(用反向 DNS,会慢一些)
python scripts/backfill_vendor.py --with-hostname
# 5. 加 --yes 跳过确认提示(cron 用)
python scripts/backfill_vendor.py --yes
# 6. 限制处理数量(测试用)
python scripts/backfill_vendor.py --dry-run --limit 100
```
### 参数
| 参数 | 说明 |
|------|------|
| `--dry-run` | 只统计,不写库 |
| `--with-hostname` | 同时回填 hostnameDNS 解析) |
| `--timeout N` | DNS 解析超时秒数(默认 2) |
| `--limit N` | 最多处理 N 条,0 = 不限 |
| `--yes` / `-y` | 跳过确认提示 |
### 常见场景
**场景 1:升级后 IP 厂商显示 -**
```bash
pip install mac-vendor-lookup # 装新依赖
python scripts/backfill_vendor.py --dry-run # 先看数量
python scripts/backfill_vendor.py # 实际回填
```
**场景 2:扫描能拿到 MAC,但扫描时 vendor 字段没写**
说明扫描代码逻辑有 bug,应该改 `enhanced_scan_service.py``update_ip_from_scan_result()` 而不是用本脚本。脚本只是数据修复。
**场景 3:所有 IP 都没有 MAC**
说明扫描本身没成功,根因在网络层(防火墙、扫描不到、ICMP/ARP 都被禁)。需要先解决扫描链路问题。
### 依赖
- `mac-vendor-lookup`IEEE OUI 官方数据库,约 17 万条记录)
- 项目本身的依赖(FastAPI、SQLAlchemy、PyMySQL 等)
- 可选:`pynmblookup`NetBIOS 主机名发现,仅 Windows 设备需要)
### 注意事项
- 脚本会**直接写库**,建议先 dry-run
- hostname 反向 DNS 可能比较慢(取决于目标设备是否配置了 PTR 记录)
- 不会改写已经存在的 vendor/hostname(只填空值)
- 不影响扫描逻辑本身(不会自动重新扫描)
### 故障排查
如果脚本运行时报 `ImportError: No module named 'app'`,说明 `sys.path` 没找到 backend 目录。在 backend 目录里执行:
```bash
cd /root/ipam/backend
python scripts/backfill_vendor.py
```
如果脚本运行后 `未识别 = 候选 IP 总数`,说明 mac_vendor_lookup 库没装或加载失败:
```bash
pip show mac-vendor-lookup # 检查库是否装好
python -c "from mac_vendor_lookup import MacLookup; print('OK')" # 测试导入
```
如果 OUI 数据库太旧(库自带的 IEEE OUI 文件是 2022 年的快照),新厂商识别不到,可以更新:
```bash
pip install -U mac-vendor-lookup
# 或下载最新 IEEE OUI 文件
python -c "from mac_vendor_lookup import MacLookup; MacLookup().update_vendors()"
```
+177
View File
@@ -0,0 +1,177 @@
#!/usr/bin/env python3
"""
回填 IP 厂商信息(vendor
对所有 mac_address 有值但 vendor 为空的 IP,调用 EnhancedScanService.get_mac_vendor
回填厂商名。基于 IEEE OUI 官方数据库(mac-vendor-lookup 包),无需重新扫描。
用法:
# 先在 git pull 之后确保依赖装了
pip install mac-vendor-lookup
# 干跑(只看不写)
python scripts/backfill_vendor.py --dry-run
# 实际写入
python scripts/backfill_vendor.py
# 同时回填主机名(反向 DNS 解析)
python scripts/backfill_vendor.py --with-hostname
参数:
--dry-run 只统计能识别的 IP 数量,不写库
--with-hostname 同时回填 hostname(需要 DNS 解析,会比较慢)
--timeout N DNS 解析超时秒数(默认 2)
--limit N 最多处理 N 条(用于测试)
--yes 跳过确认提示
"""
import sys
import os
import time
import argparse
from datetime import datetime
# 让脚本能 import app.* 模块
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
BACKEND_DIR = os.path.dirname(SCRIPT_DIR)
sys.path.insert(0, BACKEND_DIR)
def parse_args():
p = argparse.ArgumentParser(
description="回填 IP 表的 vendor / hostname 字段",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=__doc__,
)
p.add_argument("--dry-run", action="store_true", help="只统计,不写库")
p.add_argument("--with-hostname", action="store_true", help="同时回填主机名(反向 DNS 解析)")
p.add_argument("--timeout", type=int, default=2, help="DNS 解析超时秒数(默认 2")
p.add_argument("--limit", type=int, default=0, help="最多处理 N 条,0 表示不限")
p.add_argument("--yes", "-y", action="store_true", help="跳过确认提示")
return p.parse_args()
def confirm(prompt: str, skip: bool) -> bool:
if skip:
return True
try:
ans = input(f"{prompt} [y/N] ").strip().lower()
return ans in ("y", "yes")
except EOFError:
return False
def main():
args = parse_args()
# 延迟 import,确保 sys.path 设好
from app.core.database import SessionLocal
from app.models.network import IPAddress
from app.services.enhanced_scan_service import EnhancedScanService
db = SessionLocal()
try:
# 找出 mac_address 有值但 vendor 为空的 IP
query = db.query(IPAddress).filter(
IPAddress.mac_address.isnot(None),
IPAddress.vendor.is_(None),
)
if args.limit > 0:
query = query.limit(args.limit)
candidates = query.all()
total_candidates = len(candidates)
print(f"[扫描] 找到 {total_candidates} 个 mac_address 有值但 vendor 为空的 IP")
if total_candidates == 0:
print("✅ 没有需要回填的 IP")
return 0
# 统计
filled_vendor = 0
skipped_vendor = 0
filled_hostname = 0
skipped_hostname = 0
errors = 0
start = time.time()
for i, ip in enumerate(candidates, 1):
mac = ip.mac_address
vendor = EnhancedScanService.get_mac_vendor(mac)
if vendor:
if not args.dry_run:
ip.vendor = vendor
filled_vendor += 1
else:
skipped_vendor += 1
# 反向 DNS(可选)
hostname = None
if args.with_hostname:
try:
hostname = EnhancedScanService.reverse_dns_lookup(ip.ip_address, timeout=args.timeout)
except Exception as e:
print(f" [WARN] {ip.ip_address} DNS 解析失败: {e}")
errors += 1
if hostname:
if not args.dry_run:
ip.hostname = hostname
filled_hostname += 1
else:
skipped_hostname += 1
# 每 100 条打印进度
if i % 100 == 0 or i == total_candidates:
elapsed = time.time() - start
rate = i / elapsed if elapsed > 0 else 0
eta = (total_candidates - i) / rate if rate > 0 else 0
print(
f" 进度: {i}/{total_candidates} "
f"已识别 vendor={filled_vendor}, 未识别={skipped_vendor} "
f"elapsed={elapsed:.1f}s, ETA={eta:.1f}s"
)
# 提交(如果非 dry-run
if not args.dry_run:
if not confirm(f"确认写入数据库?将更新 {filled_vendor} 个 IP 的 vendor 字段"
+ (f" + {filled_hostname} 个 IP 的 hostname 字段" if args.with_hostname else ""),
args.yes):
print("已取消")
db.rollback()
return 1
db.commit()
print()
print("✅ 已提交数据库")
else:
print()
print(f"[DRY-RUN] 不写库;如需执行去掉 --dry-run")
# 汇总
print()
print("=== 汇总 ===")
print(f" 候选 IP 总数: {total_candidates}")
print(f" 识别到厂商: {filled_vendor}")
print(f" 未识别(OUI 未收录或格式异常): {skipped_vendor}")
if args.with_hostname:
print(f" 解析到主机名: {filled_hostname}")
print(f" DNS 解析失败或超时: {skipped_hostname}")
print(f" 异常次数: {errors}")
return 0
except KeyboardInterrupt:
print("\n[中断] 已回滚未提交的修改")
db.rollback()
return 130
except Exception as e:
print(f"\n[ERROR] {e}")
import traceback
traceback.print_exc()
db.rollback()
return 1
finally:
db.close()
if __name__ == "__main__":
sys.exit(main())
+11 -2
View File
@@ -63,20 +63,29 @@ export const scanApi = {
return api.post(`/scan/ping/${id}`)
},
// 综合扫描网段
// options: { enable_ping, enable_arp, enable_dns, timeout }
// options: { enable_ping, enable_arp, enable_dns, enable_netbios, timeout }
// 单次调用覆盖全局 30s 超时,因为 /24 网段扫描要数分钟
comprehensiveScan(id, options = {}) {
const {
enable_ping = true,
enable_arp = true,
enable_dns = true,
enable_netbios = true,
timeout = 2
} = options
return api.post(`/scan/comprehensive/${id}`, {}, {
params: { enable_ping, enable_arp, enable_dns, timeout },
params: { enable_ping, enable_arp, enable_dns, enable_netbios, timeout },
timeout: 600000 // 10 分钟,足以应付 /16 级别网段
})
},
// 批量补全厂商信息
batchFillVendor(params) {
return api.post('/scan/batch/vendor', {}, { params })
},
// 批量发现主机名
batchDiscoverHostname(params) {
return api.post('/scan/batch/hostname', {}, { params })
},
// 单个IP扫描
scanIP(ipAddress) {
return api.post(`/scan/comprehensive/ip/${ipAddress}`)
+23
View File
@@ -8,6 +8,10 @@
<el-icon><Search /></el-icon>
批量扫描
</el-button>
<el-button type="warning" @click="batchFillVendor" :loading="vendorLoading">
<el-icon><Refresh /></el-icon>
补全厂商
</el-button>
</div>
</template>
@@ -176,6 +180,7 @@ const route = useRoute()
const loading = ref(false)
const submitting = ref(false)
const scanning = ref(false)
const vendorLoading = ref(false)
const dialogVisible = ref(false)
const tableData = ref([])
@@ -278,6 +283,24 @@ const startBatchScan = async () => {
}
}
const batchFillVendor = async () => {
if (!filters.networkId) {
ElMessage.warning('请先选择网段')
return
}
vendorLoading.value = true
try {
const res = await scanApi.batchFillVendor({ network_id: filters.networkId })
ElMessage.success(res.message)
loadIPs()
} catch (error) {
ElMessage.error('补全厂商失败')
console.error('补全厂商错误:', error)
} finally {
vendorLoading.value = false
}
}
const editIP = (row) => {
ipForm.id = row.id
ipForm.ip_address = row.ip_address