在企业内网、校园网络或小型机房中,管理员常需实时掌握服务器、网关等设备的在线状态与延迟。本文实现一个基于Python的全栈网络监控工具,后端使用FastAPI + SQLAlchemy + SQLite,前端采用Vue 3 + Vite,具备用户认证、设备CRUD、ICMP/TCP探测、指标记录和告警处理功能。项目结构完整,可作为运维监控平台的基础版本。
技术栈
后端:FastAPI、SQLAlchemy、Pydantic、SQLite
前端:Vue 3、Vite、Fetch API、组件化开发
认证:PBKDF2密码哈希、Bearer Token、Token吊销
网络探测:优先ICMP Ping,失败时回退TCP 80端口
数据库模型设计
使用SQLite,核心表包括:
- users:用户表,字段username、email、hashed_password
- access_tokens:登录令牌表,字段token、user_id、revoked
- devices:监控设备表,字段name、ip_address、last_status、owner_id
- metrics:检测指标表,字段device_id、status、latency_ms、checked_at
- alerts:告警表,字段device_id、level、message、resolved
User与Device为一对多关系,Device与Metric、Alert也是一对多,实现用户数据隔离。
核心模型示例(SQLAlchemy ORM):- from sqlalchemy import Column, Integer, String, Text, Boolean, Float, DateTime, ForeignKey
- from sqlalchemy.ext.declarative import declarative_base
- Base = declarative_base()
- class Device(Base):
- __tablename__ = "devices"
- id = Column(Integer, primary_key=True, index=True)
- name = Column(String(80), nullable=False)
- ip_address = Column(String(64), nullable=False)
- description = Column(Text, default="")
- check_interval = Column(Integer, default=60)
- is_active = Column(Boolean, default=True)
- last_status = Column(String(20), default="unknown")
- last_latency_ms = Column(Float, nullable=True)
- last_checked_at = Column(DateTime, nullable=True)
- owner_id = Column(Integer, ForeignKey("users.id"), nullable=False)
复制代码
后端API设计
所有接口以/api开头,注册、登录、健康检查无需认证,其他接口需携带Bearer Token。关键接口:
- GET /api/health
- POST /api/auth/register
- POST /api/auth/login
- POST /api/auth/logout(需登录)
- GET /api/users/me(需登录)
- GET /api/dashboard(需登录)
- GET/POST /api/devices(需登录)
- PUT/DELETE /api/devices/{device_id}(需登录)
- POST /api/devices/{device_id}/check(需登录)
- GET /api/devices/{device_id}/metrics(需登录)
- GET /api/alerts(需登录)
- POST /api/alerts/{alert_id}/resolve(需登录)
认证依赖通过FastAPI Depends实现,自动从请求头解析Authorization: Bearer <token>,无效或吊销时返回401。
核心代码解析
1. 密码哈希与Token生成
不存储明文密码,使用PBKDF2加盐哈希:- import os
- import hashlib
- def hash_password(password: str) -> str:
- salt = os.urandom(16).hex()
- digest = hashlib.pbkdf2_hmac("sha256", password.encode(), salt.encode(), 120_000).hex()
- return f"pbkdf2_sha256$120000${salt}${digest}"
复制代码 登录成功后生成随机Token写入access_tokens表,退出时将Token标记为revoked=True。
2. 受保护接口的依赖注入
FastAPI依赖从Token中获取当前用户:- def get_current_user(token: str = Depends(extract_token), db: Session = Depends(get_db)):
- user = crud.get_user_by_token(db, token)
- if not user:
- raise HTTPException(status_code=401, detail="Token 无效或已过期")
- return user
复制代码 接口只需添加current_user: models.User = Depends(get_current_user)即可鉴权。
3. 网络检测逻辑
检测设备时优先执行ICMP Ping,失败后尝试TCP 80端口探测:- import subprocess
- def probe_device(ip_address: str):
- # ICMP Ping,超时2秒
- result = subprocess.run(["ping", "-c", "1", "-W", "2", ip_address], capture_output=True, timeout=3)
- if result.returncode == 0:
- # 解析延迟(需要从输出提取,此处省略细节)
- return ("online", latency, "ICMP Ping 成功")
- # 失败后TCP探测
- ok, latency, message = _tcp_probe(ip_address) # 尝试连接TCP 80
- return ("online", latency, message) if ok else ("offline", None, message)
复制代码 检测完成后更新设备状态,写入metrics表;设备离线时自动生成critical级别告警。
4. 前端鉴权请求封装
前端src/api.js统一管理API调用,自动携带Token:- async function request(path, options = {}) {
- const headers = { 'Content-Type': 'application/json', ...(options.headers || {}) }
- const token = getToken()
- if (token) headers.Authorization = `Bearer ${token}`
- const response = await fetch(`${API_BASE}${path}`, { ...options, headers })
- if (!response.ok) throw new Error('请求失败')
- return response.json()
- }
复制代码 所有业务接口通过request调用,无需重复处理认证。
部署与运行
后端启动(假设项目目录为project/):- cd backend
- python3 -m venv .venv
- source .venv/bin/activate
- pip install -r requirements.txt
- uvicorn app.main:app --reload --host 0.0.0.0 --port 8000
复制代码 前端启动:- cd frontend
- npm install
- npm run dev
复制代码 使用流程:打开前端页面注册登录 → 添加设备(如8.8.8.8) → 点击“立即检测”查看状态 → 在告警中心处理离线告警。
可扩展方向:定时巡检、邮件/企业微信告警、SNMP采集、Prometheus指标导出、Grafana可视化等。该项目适合学习Python全栈开发、FastAPI接口设计、Vue 3前端交互及运维监控系统原型设计。 |