ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

Python手撸入侵检测系统:从Raw Socket到SQLi/SYN Flood检测

Python手撸入侵检测系统:从Raw Socket到SQLi/SYN Flood检测 简介本资源是一套基于Python实现的网络入侵检测与防御系统完整毕业设计项目面向计算机相关专业本科生及网络安全初学者解决课程设计、期末大作业与毕业设计选题难、代码调试难、文档撰写难等实际问题。压缩包共38个文件含14个核心Python源码如app.py、routes.py、install.py、Docker部署配置Dockerfile、docker-compose.yml、前端页面HTML/JS/CSS、配置文件YML/JSON及说明文档README.md、requirements.txt整体仅97KB轻量易部署。已有63人下载学习项目经导师指导并获99分高分评价代码结构清晰、注释完整、依赖明确小白可直接运行配套文档涵盖系统架构、模块功能说明、环境搭建步骤与常见问题排错提示支持快速复现与二次开发。1. 这不是写个 ping 脚本就交差的毕业设计一个能真抓到 SYN Flood、SQL 注入和横向移动痕迹的 Python 入侵检测系统到底要填多少坑才能跑通你手里的“基于Python的网络入侵检测与防御系统”毕业设计大概率不是在 Wireshark 里点开 pcap 文件、标几条红色告警就完事。真实场景里它得在 Ubuntu 22.04 上持续监听 eth0 接口把混杂模式下捕获的原始字节流实时解析成 IP/TCP/HTTP 层结构识别出异常 TCP 标志组合比如 SYNFINRST 同时置位、HTTP 请求体里的 OR 11--特征、以及内网中某台 Windows 主机在凌晨 2:17 向 50 台其他主机发起 SMB 连接的横向移动行为还得在检测到高危事件后自动调用 iptables 命令封禁源 IP并把完整上下文时间戳、五元组、载荷前 128 字节、规则匹配路径写入 SQLite 数据库供后续回溯——而这一切必须在不依赖任何商业 IDS 引擎、纯靠 Python 标准库 Scapy Pandas Flask 搭建的前提下稳定运行 72 小时以上。这不是课程实验是答辩老师会连着追问“你这个阈值怎么定的”“误报率测过没”“如果攻击者分片发送 SQL 注入 payload 怎么办”的硬核交付物。适合计算机/网络安全专业、已掌握 Python 基础语法、Linux 命令行、TCP/IP 协议栈但还没真正把“检测”二字从概念落到socket.recvfrom()和scapy.layers.inet.TCP.flags字段上的同学。2. 从 raw socket 到可落地的检测流水线为什么不用 Snort、Suricata而坚持手撸核心模块2.1 为什么毕业设计必须自己实现协议解析层——绕不开的三个硬约束很多同学第一反应是“直接调用 Snort 不香吗”——香但毕设不许。毕业设计的核心考察点从来不是“谁家轮子好”而是“你能不能理解轮子怎么转”。具体到本项目有三个不可绕过的硬约束可解释性要求答辩时老师问“你这条规则为什么能检出 SQLi”你不能答“因为 Snort 规则库里写了”。你得能指着if bSELECT in payload and bFROM in payload and bWHERE in payload这行代码说明它如何应对大小写混淆sElEcT、空格替换%20、/**/、注释符截断--vs#等基础绕过环境可控性要求学校实验室服务器通常只开放基础 Python 环境禁止安装 C 编译工具链或 root 权限服务。Snort/Suricata 需要apt install snort 配置/etc/snort/ 启动守护进程而本方案仅需pip install scapy pandas flask apscheduler所有逻辑在用户态 Python 进程内完成教学完整性要求从数据链路层Ethernet 帧头→ 网络层IP 头校验和验证→ 传输层TCP 序列号/窗口大小分析→ 应用层HTTP 方法/URI/Body 提取每一层的字段偏移、长度计算、字节序转换都是网络原理课的具象化落点。跳过这一环等于把《计算机网络》课本最后三章撕了交作业。提示这不是反对用成熟引擎而是明确毕业设计的定位——它是你技术能力的“黑匣子解剖报告”不是“集成方案说明书”。2.2 构建最小可行检测流水线四层模块拆解与数据流向整个系统按数据处理阶段划分为四个核心模块数据单向流动无状态耦合模块名输入输出关键技术点是否必须自研捕获层Capture网卡原始字节流scapy.packet.Packet对象流socket.AF_PACKETsocket.SOCK_RAWsetsockopt(SO_ATTACH_FILTER)BPF 过滤器✅ 必须绕过内核协议栈解析层ParsePacket 对象结构化字典{src_ip: 192.168.1.100, dst_port: 80, http_method: POST, http_uri: /login.php, payload: busernameadminpass...}scapy.layers.inet.*字段提取 bytes.decode(utf-8, errorsignore)容错✅ 必须理解协议字段语义检测层Detect解析后字典告警字典{level: CRITICAL, rule_id: SQLI_001, match_detail: Found UNION SELECT in POST body}正则预编译re.compile(rUNION\sSELECT, re.I) 滑动窗口统计SYN 包速率 状态机HTTP 请求/响应配对✅ 必须体现检测逻辑设计响应层Response告警字典封禁命令执行结果 数据库存储记录 Web 页面实时刷新subprocess.run([iptables, -A, INPUT, -s, src_ip, -j, DROP])sqlite3.connect().execute(INSERT ...)⚠️ 响应动作可选但接口必须预留这个流水线不是理论模型是代码级的强制约定。每个模块输出必须是下一个模块的明确输入格式杜绝“我这里 return 了个 list你那边自己 for 循环拆吧”这类模糊契约。2.3 捕获层实操用 raw socket 绕过内核协议栈精准截获原始流量关键不在“能抓包”而在“只抓需要的包、且不被内核干扰”。常见错误是直接用scapy.sniff()它底层仍走libpcap会触发内核 TCP 重传、ACK 生成等行为导致你看到的不是线路上的真实帧而是内核加工后的“二手数据”。正确做法是直连网卡驱动用AF_PACKET创建 raw socket手动设置 BPF 过滤器import socket import struct import ctypes def create_raw_socket(interfaceeth0): # 创建 AF_PACKET socket直接对接网卡驱动 sock socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.htons(0x0003)) # 绑定到指定网卡注意interface 名必须真实存在用 ip link show 查 sock.bind((interface, 0)) # 设置 BPF 过滤器只收 IPv4 TCP/UDP 包且目的端口为 80/443/22/3306Web/SSH/MySQL # BPF 汇编指令此处用预编译字节码避免 runtime 编译开销 # 指令含义ldh [12] → 读以太网类型jeq #0x800, ..., ... → 跳转到 IPv4 分支 # ldxb 4*([14]0xf) → 计算 IP 头长ldb [x 9] → 读 IP 协议字段jeq #6, ..., ... → TCP # ldh [x 14] → 读 TCP 目的端口jeq #80, ..., ... → HTTP bpf_filter bytes([ 0x28, 0x00, 0x00, 0x00, # ldh [0] 0x15, 0x00, 0x00, 0x00, 0x00, 0x00, 0x08, 0x00, # jeq #0x800, continue, drop 0x30, 0x00, 0x00, 0x00, # ldb [0] 0x15, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x06, # jeq #6, continue, drop (TCP) 0x28, 0x00, 0x00, 0x10, # ldh [16] (TCP dst port) 0x15, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x50, # jeq #80, accept, drop 0x15, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0xbb, # jeq #443, accept, drop 0x15, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x16, # jeq #22, accept, drop 0x15, 0x00, 0x00, 0x00, 0x00, 0x00, 0x0c, 0xa6, # jeq #3306, accept, drop 0x06, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, # ret #0 (drop all others) 0x06, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, # ret #0 (accept) ]) # 加载 BPF 过滤器到 socket sock.setsockopt(socket.SOL_SOCKET, socket.SO_ATTACH_FILTER, struct.pack(HL, len(bpf_filter)//8, ctypes.addressof(ctypes.c_char_p(bpf_filter).contents))) return sock # 使用示例 sock create_raw_socket(eth0) while True: packet, _ sock.recvfrom(65535) # 直接获取原始字节流 # 后续交给 scapy 解析pkt Ether(packet)参数说明与踩坑点socket.AF_PACKET是 Linux 特有Windows 下需换AF_INETSOCK_RAW但无法捕获入站包仅限发包BPF 过滤器字节码必须严格按 8 字节对齐否则SO_ATTACH_FILTER调用失败并抛OSError: Invalid argumentrecvfrom(65535)的缓冲区大小必须 ≥ 最大传输单元MTU否则分片包被截断导致 TCP 重组失败bind((interface, 0))中的0表示接收所有协议类型若只想要 IP 包可改用socket.htons(0x0800)。这一步做完你拿到的就是网线上最原始的字节流没有内核 TCP 栈的任何干预——这是后续所有检测逻辑可信的前提。3. 解析层与检测层如何让 Python 看懂 TCP 标志位、HTTP 头、SQL 注入特征3.1 解析层从 raw bytes 到结构化字段Scapy 的正确打开方式Scapy 强大但新手常犯两个致命错误一是pkt.show()看到字段就以为能直接pkt[TCP].flags访问二是忽略分片重组。我们逐层拆解from scapy.all import Ether, IP, TCP, UDP, Raw, HTTPRequest, HTTPResponse import re def parse_packet(raw_bytes): 将 raw socket 收到的字节流解析为结构化字典 返回 None 表示解析失败如非 IP 包、校验和错误 try: # Step 1: 解析以太网帧必须否则无法判断上层协议 ether Ether(raw_bytes) if not ether.haslayer(IP): return None ip_layer ether[IP] # Step 2: IP 层校验和验证毕业设计必须体现协议细节 # 手动计算校验和清零校验和字段按 16-bit 求和取反 ip_header bytes(ip_layer)[:20] # IP 头固定 20 字节 checksum_calc 0 for i in range(0, len(ip_header), 2): if i 1 len(ip_header): word (ip_header[i] 8) ip_header[i1] checksum_calc word checksum_calc (checksum_calc 0xFFFF) (checksum_calc 16) checksum_calc ~checksum_calc 0xFFFF if checksum_calc ! ip_layer.chksum: return None # 校验失败丢弃 # Step 3: 提取传输层和应用层信息 result { src_ip: ip_layer.src, dst_ip: ip_layer.dst, proto: ip_layer.proto, timestamp: time.time(), payload: b # 初始化 } # TCP 处理 if ip_layer.proto 6 and ip_layer.haslayer(TCP): tcp_layer ip_layer[TCP] result.update({ src_port: tcp_layer.sport, dst_port: tcp_layer.dport, tcp_flags: tcp_layer.flags, # 这是整数如 0x12 SYNACK seq_num: tcp_layer.seq, ack_num: tcp_layer.ack, }) # 提取 TCP 载荷HTTP/HTTPS/自定义协议 if tcp_layer.haslayer(Raw): result[payload] bytes(tcp_layer[Raw]) # 尝试解析 HTTP仅限明文 if tcp_layer.dport 80 or tcp_layer.sport 80: try: # 用正则粗略提取 HTTP 方法和 URI避免 full HTTP parser 依赖 http_match re.match(b(GET|POST|PUT|DELETE) ([^ ]) HTTP/[^\\r\\n], result[payload]) if http_match: result[http_method] http_match.group(1).decode(ascii) result[http_uri] http_match.group(2).decode(utf-8, errorsignore) except: pass # UDP 处理DNS、SNMP 等 elif ip_layer.proto 17 and ip_layer.haslayer(UDP): udp_layer ip_layer[UDP] result.update({ src_port: udp_layer.sport, dst_port: udp_layer.dport, }) if udp_layer.haslayer(Raw): result[payload] bytes(udp_layer[Raw]) return result except Exception as e: # 解析异常不抛出返回 None由上层统一丢弃 return None # 使用示例 raw_pkt sock.recvfrom(65535)[0] parsed parse_packet(raw_pkt) if parsed and parsed.get(http_method) POST: print(fDetected POST to {parsed.get(http_uri, )})关键逻辑说明Ether(raw_bytes)是必须的第一步因为 raw socket 收到的是包含以太网头的完整帧不解析就无法知道上层是 IP 还是 ARPIP 校验和验证是毕业设计的加分项体现你理解协议规范代码中展示了手动计算过程tcp_layer.flags返回的是整数如0x12不是字符串SYNACK后续检测需用位运算判断(flags 0x02) ! 0表示 SYN 置位HTTP 解析用正则而非scapy.layers.http.HTTPRequest因为后者在分片、乱序、TLS 加密场景下极易崩溃而毕业设计只需处理明文 HTTP 测试流量。3.2 检测层三条核心规则的 Python 实现与绕过对抗思路检测不是简单if union select in payload。我们实现三个典型场景每条都带绕过对抗设计▶ 规则 1SYN Flood 检测网络层 DDoSfrom collections import defaultdict, deque import time class SynFloodDetector: def __init__(self, window_sec10, threshold100): self.window_sec window_sec self.threshold threshold # 按源 IP 统计 SYN 包数量{src_ip: deque([(timestamp, count)])} self.syn_counts defaultdict(lambda: deque(maxlen1000)) def update(self, pkt_dict): if pkt_dict.get(proto) 6: # TCP flags pkt_dict.get(tcp_flags, 0) if (flags 0x02) ! 0: # SYN bit set src_ip pkt_dict[src_ip] now time.time() self.syn_counts[src_ip].append(now) # 计算过去 window_sec 内的 SYN 数量 window_start now - self.window_sec recent_syns [t for t in self.syn_counts[src_ip] if t window_start] if len(recent_syns) self.threshold: return { level: CRITICAL, rule_id: SYN_FLOOD_001, match_detail: fSYN packets from {src_ip}: {len(recent_syns)} in {self.window_sec}s } return None # 初始化检测器window10s, threshold100 syn_detector SynFloodDetector(window_sec10, threshold100)绕过对抗思考攻击者可能降低发包速率如 90pps或分散 IPbotnet。对策是增加“连接完成率”指标统计SYN与SYNACK的比例若长期低于 5%即判定为扫描或 Flood。▶ 规则 2SQL 注入检测应用层攻击import re class SqlInjectionDetector: def __init__(self): # 预编译正则提升性能毕业设计必须考虑效率 self.patterns [ # UNION SELECT 绕过大小写、空格替换、注释符 re.compile(rb(?i)union\sselect, re.DOTALL), re.compile(rb(?i)select\s.*?\sfrom, re.DOTALL), re.compile(rb(?i)insert\sinto\s\w\svalues, re.DOTALL), # 常见注释符绕过 re.compile(rb--\s*[^\r\n]*, re.DOTALL), re.compile(rb#\s*[^\r\n]*, re.DOTALL), # 编码绕过URL 编码、HEX 编码 re.compile(rb%20|%09|%0a|%0d, re.DOTALL), # 空格/Tab/换行编码 ] def detect(self, payload): if not isinstance(payload, bytes) or len(payload) 10: return None for i, pattern in enumerate(self.patterns): if pattern.search(payload): # 记录匹配到的 pattern ID便于调试 return { level: HIGH, rule_id: fSQLI_PATTERN_{i1:03d}, match_detail: fMatched pattern #{i1}: {pattern.pattern[:50]} } return None sql_detector SqlInjectionDetector()绕过对抗思考正则无法覆盖所有变种如SEL/*abc*/ECT但毕业设计目标是“可演示、可解释”。重点在于① 用re.DOTALL让.匹配换行② 用(?i)全局忽略大小写③ 限制payload长度避免 OOM。▶ 规则 3横向移动检测内网行为分析from collections import defaultdict import time class LateralMovementDetector: def __init__(self, window_sec300, threshold_hosts5): self.window_sec window_sec self.threshold_hosts threshold_hosts # 按源 IP 统计其访问的目标 IP 列表{src_ip: [(timestamp, dst_ip)]} self.connections defaultdict(list) def update(self, pkt_dict): src_ip pkt_dict.get(src_ip) dst_ip pkt_dict.get(dst_ip) dst_port pkt_dict.get(dst_port) # 只关注高危端口SMB(445), RDP(3389), SSH(22), WinRM(5985) if dst_port not in [445, 3389, 22, 5985]: return None now time.time() self.connections[src_ip].append((now, dst_ip)) # 清理过期记录 window_start now - self.window_sec self.connections[src_ip] [ (t, ip) for t, ip in self.connections[src_ip] if t window_start ] # 统计该源 IP 在窗口内访问的不同目标 IP 数量 unique_dst_ips set(ip for _, ip in self.connections[src_ip]) if len(unique_dst_ips) self.threshold_hosts: return { level: MEDIUM, rule_id: LATERAL_MOVE_001, match_detail: f{src_ip} accessed {len(unique_dst_ips)} hosts via port {dst_port} in {self.window_sec}s } return None lateral_detector LateralMovementDetector(window_sec300, threshold_hosts5)绕过对抗思考攻击者可能用代理或慢速扫描。对策是增加“时间分布熵”分析若 5 个连接在 5 秒内均匀发出熵低比随机分布在 5 分钟内熵高更可疑。4. 避坑指南毕业答辩前必须亲手踩过的 5 个深坑及血泪解决方案4.1 现象程序运行 2 小时后 CPU 占用飙升至 100%top显示 Python 进程吃满一个核原因scapy.sniff()或socket.recvfrom()在无包到达时持续轮询busy-wait未加timeout或select()阻塞。解决若用socket必须设置sock.settimeout(1.0)并在except socket.timeout:中做空转处理若用scapy改用sniff(prncallback, timeout1, store0)store0禁止保存包到内存毕业设计必改在主循环中加入time.sleep(0.001)虽微小但可降 CPU 至 5% 以下。4.2 现象检测到 SQL 注入告警但 payload 字段显示为乱码如b\xff\xd8\xff\xe0\x00\x10...原因payload是原始 TCP 载荷可能包含二进制数据图片、加密流量直接decode(utf-8)报错后被忽略导致后续正则匹配失效。解决永远用bytes类型做正则匹配不要先 decode若需日志显示用payload[:100].hex()或repr(payload[:100])在parse_packet()中对payload做errorsignore的安全 decode仅用于展示不用于检测。4.3 现象iptables 封禁命令执行成功但iptables -L查不到新规则或封禁后攻击仍在继续原因subprocess.run([iptables, -A, INPUT, ...])添加的是INPUT链但攻击包可能走FORWARD链如本机作网关或规则被更早的ACCEPT规则提前匹配。解决毕业设计统一用INPUT链并确保在iptables规则链顶部插入[iptables, -I, INPUT, 1, -s, src_ip, -j, DROP]执行后立即验证subprocess.run([iptables, -C, INPUT, -s, src_ip, -j, DROP], checkFalse)重要提示答辩前务必iptables -F清空规则避免残留规则干扰测试。4.4 现象多线程处理包时SQLite 数据库报database is locked错误告警丢失原因多个线程同时INSERT INTO alertsSQLite 默认串行化写入后到的线程等待超时。解决毕业设计不推荐多线程写 DB改用单线程队列主线程捕获 → 生产者线程解析/检测 →queue.Queue()传递告警 → 消费者线程唯一批量写入或改用sqlite3.connect(..., timeout10)但 timeout 过长影响实时性终极方案用threading.Lock()包裹cursor.execute()简单有效。4.5 现象在虚拟机VMware/VirtualBox中测试scapy无法捕获入站包或AF_PACKETsocket 创建失败原因虚拟网卡驱动不支持AF_PACKET或混杂模式未开启。解决VMware编辑虚拟机设置 → 网络适配器 → 高级 → 勾选 “混杂模式接受”VirtualBoxVBoxManage modifyvm VMName --nicpromisc2 allow-all备用方案改用AF_INETSOCK_RAW仅限发包或直接在物理机 Ubuntu 22.04 上开发强烈推荐避坑率 90%。5. 响应层与可视化如何让答辩老师一眼看懂你的系统在“动”5.1 响应层从封禁命令到可审计的日志闭环检测到攻击只是开始响应动作必须可验证、可审计、可撤回。毕业设计不求全自动但接口必须健壮import subprocess import sqlite3 import time from datetime import datetime class ResponseEngine: def __init__(self, db_pathalerts.db): self.db_path db_path self._init_db() def _init_db(self): 初始化 SQLite 数据库建表 conn sqlite3.connect(self.db_path) conn.execute( CREATE TABLE IF NOT EXISTS alerts ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp TEXT NOT NULL, level TEXT NOT NULL, rule_id TEXT NOT NULL, src_ip TEXT, dst_ip TEXT, src_port INTEGER, dst_port INTEGER, match_detail TEXT, action TEXT, action_result TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ) conn.close() def execute_block(self, src_ip): 执行 iptables 封禁并记录到数据库 try: # 插入数据库先记再执行避免执行失败导致日志缺失 conn sqlite3.connect(self.db_path) cursor conn.cursor() cursor.execute( INSERT INTO alerts (timestamp, level, rule_id, src_ip, action, action_result) VALUES (?, ?, ?, ?, ?, ?), (datetime.now().isoformat(), CRITICAL, BLOCK_IP, src_ip, iptables DROP, pending) ) alert_id cursor.lastrowid conn.commit() # 执行封禁 result subprocess.run( [iptables, -I, INPUT, 1, -s, src_ip, -j, DROP], capture_outputTrue, textTrue, timeout5 ) # 更新数据库状态 if result.returncode 0: cursor.execute( UPDATE alerts SET action_result ? WHERE id ?, (success, alert_id) ) conn.commit() return True, Blocked else: cursor.execute( UPDATE alerts SET action_result ? WHERE id ?, (ffailed: {result.stderr[:100]}, alert_id) ) conn.commit() return False, fFailed: {result.stderr[:50]} except Exception as e: return False, fException: {str(e)} finally: conn.close() # 使用 responder ResponseEngine() success, msg responder.execute_block(192.168.1.200) print(fBlock result: {success}, {msg}) # Block result: True, Blocked关键设计点先写日志再执行动作即使iptables崩溃告警记录仍在符合审计要求iptables -I INPUT 1插入到规则链顶部确保优先匹配避免被ACCEPT规则拦截timeout5防止iptables卡死影响主线程action_result字段存详细错误答辩时可导出此表证明你处理了异常。5.2 可视化用 Flask Chart.js 做一个 50 行代码的实时仪表盘毕业设计不需要炫酷 UI但必须让老师看到“系统在活”。Flask 轻量Chart.js 免费50 行搞定# app.py from flask import Flask, render_template, jsonify import sqlite3 import time app Flask(__name__) app.route(/) def dashboard(): return render_template(dashboard.html) app.route(/api/alerts) def get_alerts(): # 查询最近 10 条告警 conn sqlite3.connect(alerts.db) cursor conn.cursor() cursor.execute(SELECT * FROM alerts ORDER BY created_at DESC LIMIT 10) rows cursor.fetchall() conn.close() alerts [] for row in rows: alerts.append({ id: row[0], time: row[1][:19], # 截取 YYYY-MM-DD HH:MM:SS level: row[2], rule: row[3], src_ip: row[4], detail: row[7][:50] ... if len(row[7]) 50 else row[7], action: row[8], }) return jsonify(alerts) app.route(/api/chart-data) def chart_data(): # 统计每小时告警数模拟实际查 DB now int(time.time()) data [] for i in range(24): hour_ago now - i * 3600 # 此处应查 DBSELECT COUNT(*) FROM alerts WHERE created_at ? count 3 (i % 5) * 2 # 模拟数据 data.append(count) return jsonify(data) if __name__ __main__: app.run(host0.0.0.0, port5000, debugFalse) # debugFalse 防止生产环境暴露配套templates/dashboard.html精简版!DOCTYPE html html headtitleIDPS Dashboard/title script srchttps://cdn.jsdelivr.net/npm/chart.js/script /head body h2实时入侵检测仪表盘/h2 divcanvas idalertChart width400 height200/canvas/div h3最新告警10条/h3 table border1 trth时间/thth等级/thth规则/thth源IP/thth详情/thth动作/th/tr tbody idalertTable/tbody /table script // 加载图表 const ctx document.getElementById(alertChart).getContext(2d); new Chart(ctx, { type: line, data: { labels: Array.from({length:24}, (_,i) ${24-i}h), datasets: [{ label: 告警数/小时, data: [], borderColor: red, fill: false }] } }); // 加载告警列表 function loadAlerts() { fetch(/api/alerts) .then(r r.json()) .then(alerts { const tbody document.getElementById(alertTable); tbody.innerHTML ; alerts.forEach(a { const tr document.createElement(tr); tr.innerHTML td${a.time}/tdtd${a.level}/tdtd${a.rule}/td td${a.src_ip}/tdtd${a.detail}/tdtd${a.action}/td; tbody.appendChild(tr); }); }); } // 每5秒刷新 setInterval(() { fetch(/api/chart-data).then(r r.json()).then(data { const chart Chart.getChart(alertChart); p a hrefhttps://download.csdn.net/download/chengxuyuanlaow/90680471 stylecolor:#ec7500;font-size:14px; 本文还有配套的精品资源点击获取 /a img altmenu-r.4af5f7ec.gif srchttps://csdnimg.cn/release/wenkucmsfe/public/img/menu-r.4af5f7ec.gif stylewidth:16px;margin-left:4px;vertical-align:text-bottom;cursor:text; /p
返回列表