当前位置：首页 > news >正文

基于Django的内部网络资产发现与管理工具

news 2025/9/27 12:10:35

引言：为什么需要内部网络资产管理？

在当今数字化时代，企业内部网络环境日益复杂，各种设备（服务器、工作站、网络设备、IoT设备等）数量激增。传统的手工记录和管理方式已经无法满足现代企业的需求：

安全风险：未知设备接入可能带来安全漏洞
运维效率：手动维护资产信息耗时耗力
合规要求：各类安全审计需要完整的资产清单
业务连续性：快速定位故障设备至关重要

基于这些痛点，我们开发了这款基于Django的内部网络资产发现与管理工具，本文将深入解析其技术实现。

一、技术栈选择与项目架构

1.1 技术选型考量

在选择技术栈时，我们主要考虑以下因素：

后端框架选择：Django

成熟稳定，社区活跃
自带Admin后台，快速开发
ORM支持多种数据库
完善的安全机制

前端技术：Bootstrap + 原生JavaScript

Bootstrap提供响应式UI组件
原生JavaScript避免框架依赖
轻量级，性能优秀

网络扫描技术

Scapy：强大的网络包操作库
Socket：基础网络通信
多线程/多进程：并发处理

1.2 项目架构设计

应用层
├── Web界面 (Bootstrap + JavaScript)
├── RESTful API接口
└── 实时通信 (WebSocket)业务逻辑层
├── 资产管理模块
├── 网络扫描引擎
├── 漏洞检测模块
└── 告警处理模块数据访问层
├── Django ORM
├── 数据库抽象
└── 缓存机制基础设施层
├── 网络协议栈
├── 并发处理
└── 系统调用

二、核心功能模块详解

2.1 数据模型设计

资产管理的核心是数据模型，我们设计了以下主要模型：

Asset（资产）模型

class Asset(models.Model):STATUS_CHOICES = [('online', '在线'),('offline', '离线'),('unknown', '未知'),]name = models.CharField(max_length=100, verbose_name='资产名称')ip_address = models.GenericIPAddressField(verbose_name='IP地址')mac_address = models.CharField(max_length=17, blank=True, verbose_name='MAC地址')status = models.CharField(max_length=10, choices=STATUS_CHOICES, default='unknown')last_seen = models.DateTimeField(null=True, blank=True, verbose_name='最后在线时间')created_at = models.DateTimeField(auto_now_add=True, verbose_name='创建时间')class Meta:verbose_name = '网络资产'verbose_name_plural = '网络资产'unique_together = ['ip_address', 'mac_address']

模型关系设计

一对多：Asset → Port（一个资产有多个端口）
一对多：Asset → Vulnerability（一个资产有多个漏洞）
一对多：Asset → Alert（一个资产产生多个告警）

这种设计保证了数据的完整性和查询效率。

2.2 网络扫描引擎

网络扫描是整个系统的核心，我们实现了多种扫描策略：

2.2.1 智能网络大小估算

def _estimate_network_size(self, network):"""智能估算网络规模"""if '/' in network:# CIDR格式: 192.168.1.0/24ip_network = ipaddress.ip_network(network, strict=False)return ip_network.num_addresseselif '-' in network:# IP范围格式: 192.168.1.1-192.168.1.100start_ip, end_ip = network.split('-')start_int = self._ip_to_int(start_ip.strip())end_int = self._ip_to_int(end_ip.strip())return end_int - start_int + 1else:# 单个IPreturn 1

2.2.2 并发扫描策略

根据网络规模动态选择最优扫描方式：

def enhanced_discover_assets(self, network, scan_type='comprehensive', max_threads=50):network_size = self._estimate_network_size(network)# 智能选择扫描策略if network_size <= 50:# 小型网络：顺序扫描，避免并发开销return self._sequential_scan(network, scan_type)elif network_size <= 256:# 中型网络：线程池并发return self._thread_pool_scan(network, scan_type, min(network_size, max_threads))else:# 大型网络：进程池（绕过GIL限制）return self._process_pool_scan(network, scan_type, min(network_size // 10, 4))

2.2.3 多种扫描方式实现

ARP扫描（局域网发现）

def arp_scan(self, network):"""ARP扫描发现局域网设备"""try:# 使用Scapy发送ARP请求ans, unans = arping(network, timeout=2, verbose=0)devices = []for sent, received in ans:devices.append({'ip': received.psrc,'mac': received.hwsrc,'status': 'online'})return devicesexcept Exception as e:logger.error(f'ARP扫描失败: {e}')return []

Ping扫描（基础连通性）

def ping_scan(self, target):"""ICMP Ping扫描"""try:# 使用系统ping命令param = '-n' if os.name == 'nt' else '-c'command = ['ping', param, '1', '-w', '2', target]result = subprocess.run(command, capture_output=True, text=True)return result.returncode == 0except Exception as e:logger.error(f'Ping扫描失败: {e}')return False

2.3 视图控制器设计

2.3.1 支持AJAX和传统请求的混合架构

def mark_alert_as_read(request, pk):"""标记告警为已读 - 支持AJAX和传统请求"""try:alert = get_object_or_404(Alert, pk=pk)alert.is_read = Truealert.save()# 智能判断请求类型if request.headers.get('X-Requested-With') == 'XMLHttpRequest':# AJAX请求返回JSONreturn JsonResponse({'status': 'success','message': '告警已标记为已读','alert_id': alert.pk})else:# 传统请求重定向messages.success(request, '告警已标记为已读')return redirect('assets:alert_list')except Exception as e:logger.error(f'标记告警为已读失败: {e}')# 错误处理同样支持两种请求类型if request.headers.get('X-Requested-With') == 'XMLHttpRequest':return JsonResponse({'status': 'error','message': f'标记失败: {e}'}, status=500)else:messages.error(request, f'标记失败: {e}')return redirect('assets:alert_list')

2.3.2 批量状态检测优化

@csrf_exempt
def check_all_assets_status(request):"""批量检测所有资产状态，智能选择并发策略"""if request.method == 'POST':try:assets = list(Asset.objects.all())if not assets:return JsonResponse({'status': 'success','message': '没有资产需要检测','results': []})# 根据资产数量选择最优策略asset_count = len(assets)if asset_count <= 5:return sequential_check(assets)  # 顺序检测elif asset_count <= 20:return thread_pool_check(assets)  # 线程池else:return process_pool_check(assets) # 进程池except Exception as e:logger.error(f'批量状态检测失败: {e}')return JsonResponse({'status': 'error', 'message': str(e)})return JsonResponse({'status': 'error', 'message': '只支持POST请求'})

三、前端交互设计与实现

3.1 响应式界面设计

使用Bootstrap 5构建现代化响应式界面：

<div class="row"><div class="col-md-12"><div class="d-flex justify-content-between align-items-center mb-3"><h2>告警中心</h2><a href="{% url 'assets:mark_all_alerts_as_read' %}" class="btn btn-primary">标记所有为已读</a></div><div class="card mt-3"><div class="card-body"><div class="table-responsive"><table class="table table-striped table-hover"><!-- 表格内容 --></table></div></div></div></div>
</div>

3.2 AJAX交互实现

实现无刷新操作体验：

// 标记为已读功能
document.querySelectorAll('.mark-as-read').forEach(button => {button.addEventListener('click', function() {const alertId = this.getAttribute('data-alert-id');// 发送AJAX请求fetch(`/alerts/${alertId}/mark-as-read/`, {method: 'POST',headers: {'X-CSRFToken': '{{ csrf_token }}','Content-Type': 'application/json','X-Requested-With': 'XMLHttpRequest'}}).then(response => {if (response.ok) {return response.json().then(data => {if (data.status === 'success') {// 更新UIthis.closest('tr').querySelector('.alert-status').innerHTML = '<span class="badge bg-success">已读</span>';this.remove(); // 移除操作按钮showMessage(data.message, 'success');} else {throw new Error(data.message);}});} else {throw new Error('网络请求失败');}}).catch(error => {console.error('操作失败:', error);showMessage(`操作失败: ${error.message}`, 'error');});});
});

3.3 实时进度显示

对于长时间运行的扫描任务，提供实时进度反馈：

function startNetworkScan(network) {const progressBar = document.getElementById('scan-progress');const statusText = document.getElementById('scan-status');// 显示进度条progressBar.style.display = 'block';statusText.textContent = '扫描初始化...';// 使用WebSocket或轮询获取进度const eventSource = new EventSource('/scan-progress/');eventSource.onmessage = function(event) {const data = JSON.parse(event.data);if (data.type === 'progress') {progressBar.value = data.value;statusText.textContent = data.message;} else if (data.type === 'complete') {progressBar.value = 100;statusText.textContent = '扫描完成';eventSource.close();updateAssetList(data.assets);} else if (data.type === 'error') {statusText.textContent = `扫描错误: ${data.message}`;eventSource.close();}};// 开始扫描fetch('/enhanced_network_scan/', {method: 'POST',body: JSON.stringify({network: network})});
}

四、性能优化策略

4.1 并发处理优化

4.1.1 动态线程池管理

def _thread_pool_scan(self, network, scan_type, max_workers):"""线程池扫描实现"""import threadingfrom concurrent.futures import ThreadPoolExecutor, as_completed# 创建线程安全的计数器scanned_count = threading.Lock()completed = 0def scan_single_ip(ip):nonlocal completedtry:result = self._scan_single_target(ip, scan_type)# 线程安全更新进度with scanned_count:completed += 1logger.info(f'扫描进度: {completed}/{total_ips}')return resultexcept Exception as e:logger.error(f'扫描 {ip} 失败: {e}')return None# 获取目标IP列表targets = self._parse_network_targets(network)total_ips = len(targets)# 动态调整线程数optimal_workers = min(max_workers, total_ips, 50)  # 限制最大线程数with ThreadPoolExecutor(max_workers=optimal_workers) as executor:futures = {executor.submit(scan_single_ip, ip): ip for ip in targets}results = []for future in as_completed(futures):try:result = future.result()if result:results.append(result)except Exception as e:ip = futures[future]logger.error(f'处理 {ip} 结果时出错: {e}')return results

4.1.2 进程池优化（绕过GIL）

对于CPU密集型任务，使用进程池：

def _process_pool_scan(self, network, scan_type, max_workers):"""进程池扫描 - 适用于大型网络"""from concurrent.futures import ProcessPoolExecutor# 序列化扫描配置scan_config = {'network': network,'scan_type': scan_type,'timeout': self.timeout}# 分批处理减少进程间通信开销batch_size = max(1, len(targets) // max_workers)ip_batches = [targets[i:i + batch_size] for i in range(0, len(targets), batch_size)]with ProcessPoolExecutor(max_workers=max_workers) as executor:futures = []for batch in ip_batches:future = executor.submit(self._scan_ip_batch, batch, scan_config)futures.append(future)# 收集结果all_results = []for future in as_completed(futures):try:batch_results = future.result()all_results.extend(batch_results)except Exception as e:logger.error(f'处理批次扫描结果时出错: {e}')return all_results

4.2 数据库优化

4.2.1 批量操作减少数据库查询

def bulk_update_asset_status(self, asset_status_list):"""批量更新资产状态"""from django.db import transactiontry:with transaction.atomic():updates = []for asset_id, status, last_seen in asset_status_list:updates.append(Asset(id=asset_id,status=status,last_seen=last_seen))# 批量更新Asset.objects.bulk_update(updates, ['status', 'last_seen'])logger.info(f'批量更新了 {len(updates)} 个资产状态')except Exception as e:logger.error(f'批量更新资产状态失败: {e}')# 降级为逐个更新self._fallback_individual_update(asset_status_list)

4.2.2 查询优化

def get_assets_with_optimized_query(self):"""优化关联查询"""return Asset.objects.select_related()\.prefetch_related('port_set', 'alert_set')\.only('name', 'ip_address', 'status', 'last_seen')\.order_by('-last_seen')

4.3 内存优化

4.3.1 流式处理大型数据集

def process_large_network_scan(self, network):"""流式处理大型网络扫描"""# 分批处理，避免内存溢出batch_size = 1000all_results = []for i in range(0, total_ips, batch_size):batch_targets = targets[i:i + batch_size]batch_results = self._scan_batch(batch_targets)# 立即处理结果并释放内存processed_results = self._process_batch_results(batch_results)all_results.extend(processed_results)# 强制垃圾回收if i % (batch_size * 10) == 0:import gcgc.collect()return all_results

五、安全设计与实现

5.1 输入验证与过滤

def validate_network_input(network):"""网络输入验证"""if not network or len(network) > 50:raise ValidationError('网络参数无效')# 防止注入攻击if any(char in network for char in [';', '&', '|', '`', '$']):raise ValidationError('检测到非法字符')# 验证IP格式try:if '/' in network:ipaddress.ip_network(network, strict=False)elif '-' in network:start_ip, end_ip = network.split('-')ipaddress.ip_address(start_ip.strip())ipaddress.ip_address(end_ip.strip())else:ipaddress.ip_address(network)except ValueError:raise ValidationError('无效的IP地址或网络段')return True

5.2 权限控制

from django.contrib.auth.decorators import user_passes_testdef scanner_permission_required(view_func):"""扫描权限装饰器"""def check_scanner_permission(user):return user.is_authenticated and (user.is_staff or user.has_perm('assets.can_scan_network') oruser.groups.filter(name='NetworkScanner').exists())return user_passes_test(check_scanner_permission)(view_func)@scanner_permission_required
def network_scan_page(request):"""网络扫描页面"""# 只有授权用户才能访问return render(request, 'assets/network_scan.html')

5.3 扫描频率限制

from django.core.cache import cache
from django.http import JsonResponsedef rate_limit_scan(ip_address, max_scans_per_hour=10):"""扫描频率限制"""cache_key = f'scan_rate_limit:{ip_address}'scan_count = cache.get(cache_key, 0)if scan_count >= max_scans_per_hour:return JsonResponse({'status': 'error','message': '扫描频率过高，请稍后再试'}, status=429)# 更新计数cache.set(cache_key, scan_count + 1, 3600)  # 1小时过期return None

六、错误处理与日志系统

6.1 分层错误处理

class NetworkScannerError(Exception):"""网络扫描基础异常"""passclass ScanTimeoutError(NetworkScannerError):"""扫描超时异常"""passclass PermissionError(NetworkScannerError):"""权限异常"""passdef safe_network_scan(self, network):"""安全的网络扫描封装"""try:# 输入验证self.validate_network_input(network)# 权限检查if not self._check_scan_permission():raise PermissionError('没有扫描权限')# 执行扫描results = self.enhanced_discover_assets(network)return {'status': 'success','results': results,'scan_time': time.time() - start_time}except ValidationError as e:logger.warning(f'输入验证失败: {e}')return {'status': 'error', 'message': str(e)}except PermissionError as e:logger.warning(f'权限检查失败: {e}')return {'status': 'error', 'message': str(e)}except ScanTimeoutError as e:logger.error(f'扫描超时: {e}')return {'status': 'error', 'message': '扫描超时，请重试'}except Exception as e:logger.error(f'扫描过程异常: {e}', exc_info=True)return {'status': 'error', 'message': '扫描过程发生未知错误'}

6.2 结构化日志系统

import logging
import json
from datetime import datetimeclass JSONFormatter(logging.Formatter):"""JSON格式日志格式化器"""def format(self, record):log_entry = {'timestamp': datetime.utcnow().isoformat(),'level': record.levelname,'logger': record.name,'message': record.getMessage(),'module': record.module,'function': record.funcName,'line': record.lineno}if record.exc_info:log_entry['exception'] = self.formatException(record.exc_info)return json.dumps(log_entry, ensure_ascii=False)# 配置日志
logging.basicConfig(level=logging.INFO,format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',handlers=[logging.FileHandler('network_scanner.log'),logging.StreamHandler()]
)logger = logging.getLogger('network_scanner')

七、部署与运维

7.1 Docker容器化部署

FROM python:3.9-slim# 设置工作目录
WORKDIR /app# 复制依赖文件
COPY requirements.txt .# 安装依赖
RUN pip install --no-cache-dir -r requirements.txt# 复制应用代码
COPY . .# 创建非root用户
RUN useradd -m -u 1000 scanner
USER scanner# 暴露端口
EXPOSE 5000# 启动命令
CMD ["gunicorn", "lan_discovery.wsgi:application", "--bind", "0.0.0.0:5000"]

7.2 性能监控

import psutil
import time
from django.http import JsonResponsedef system_metrics(request):"""系统性能指标接口"""metrics = {'timestamp': time.time(),'cpu_percent': psutil.cpu_percent(interval=1),'memory_percent': psutil.virtual_memory().percent,'disk_usage': psutil.disk_usage('/').percent,'network_io': psutil.net_io_counters()._asdict(),'active_scans': cache.get('active_scan_count', 0)}return JsonResponse(metrics)

7.3 健康检查

def health_check(request):"""系统健康检查"""checks = {'database': check_database_connection(),'cache': check_cache_connection(),'scan_permission': check_scan_permission(),'disk_space': check_disk_space()}overall_status = 'healthy' if all(checks.values()) else 'unhealthy'return JsonResponse({'status': overall_status,'checks': checks,'timestamp': time.time()})