ARTICLE DETAIL

资讯详情

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

Python TCP入侵检测系统实战:从Scapy抓包到iptables自动封禁

Python TCP入侵检测系统实战:从Scapy抓包到iptables自动封禁 简介基于Python实现的TCP入侵检测系统面向网络安全方向的毕业设计、课程设计与项目开发者。系统重点解决端口扫描与Dos攻击的实时检测问题能够联动iptables完成自动防御评判逻辑综合TCP请求频率、SYN/FIN/NULL标志位比例、未开放端口请求比例等特征模型直观且易于复现。源码包共6个文件包含5个Python脚本和1个Markdown说明压缩包大小仅5KB脚本按数据嗅探、过滤分析、数据库写入、主控联动等模块拆分便于阅读和二次扩展。项目已经过功能测试可直接参考并在此基础上增加告警、可视化等能力。目前已有231人浏览学习适合需要快速搭建入侵检测原型的用户。1. TCP入侵检测系统为什么毕设选这条技术路线最稳妥被毕设题目“基于Python的TCP入侵检测系统”砸中的同学第一反应通常是去搜Snort、Suricata的配置教程结果光看规则语法就劝退一半。换个思路这个题目真正要验证的是“能不能识别异常、会不会响应止损”而不是“能不能写出内核级抓包引擎”。用Python写检测逻辑、命中后联动iptables封IP是这个题目最常见的落地形态也是课程设计、毕设演示里最出效果的做法。系统要解决的场景很具体哪台机器在试探你的端口哪个IP在给你灌SYN包检测到了能不能自动切掉攻击源。后面的阈值设计、抓包选型、防误封策略都是从这一个目标拆出来的。2. 检测核心SYN Flood与端口扫描的特征识别和阈值设计2.1 抓包方案选型为什么用Scapy而不是libpcap和Snort既然标题写死了Python抓包这条路就有三个选项直接调libpcap的Python绑定、用socket原始套接字自己解析、用Scapy。按python入门教程装好环境后大部分人会先试原始套接字方案写三层解析头、校验和代码没写多少就开始头疼。常见做法是直接用Scapy它把以太网帧、IP头、TCP头的解析封装成了现成的层对象pkt[IP].src就能取源地址pkt[TCP].flags就能看标志位不用手动算偏移量和校验和。有人会问Snort不香吗香但Snort的强项是规则库匹配它是完整的入侵检测系统对咱们这个题目来说学习成本和演示复杂度都偏高。课程设计和毕设答辩时评委更想看“检测算法是你自己想清楚的”还是“你改了一个别人的开源规则文件配置”Scapy抓包加自己写的统计逻辑接口清晰、代码可读性强答辩能讲出东西来。2.2 端口扫描怎么认从nmap的半开扫描和全连接扫描说起端口扫描的检测可以直接借鉴nmap端口扫描的思路来分析。最常见的半开扫描也就是nmap的-sS模式扫描器向目标机每个端口发SYN包如果收到SYNACK说明端口开放收到RST说明关闭。站在被扫描主机的视角看现象非常统一同一源IP在短时间内向大量不同目的端口发SYN且很多连接永远等不到第三次握手的ACK。全连接扫描-sT则会完整走完三次握手被扫描主机这边能观察到大量新建连接的尝试。检测算法可以做得简单也可以做得细。简单做法是统计“同一源IP在滑动时间窗口内访问的不同目的端口数”超过阈值就报警细一点的做法是维护一个未完成握手表SYN来了记一条同一个五元组出现ACK就移除表里积压的条目突然暴增基本可以断定有人在扫你。参考Snort的sfportscan模块思路它也是按源IP分开统计的如果两台扫描设备同时扫一台目标机按源IP分别计数才不至于在日志里只留一条记录。2.3 DoS攻击里的TCP Flood怎么认SYN Flood判定TCP型DoS攻击里最常见的是SYN Flood攻击机发大量SYN包但从不完成握手被攻击主机的半连接队列被占满合法用户连不上服务。它的特征和端口扫描不一样端口扫描是“一个源IP打很多不同端口”SYN Flood是“大量SYN包集中打同一个服务端口”或者攻击者伪造大量源IP来打同一个端口。检测逻辑上按源IP聚合SYN包数是最直观的计数器。如果一段时间窗口内同一源IP发来的SYN包数量远高于正常业务流量就判定为异常。注意阈值必须按实际环境调不能照抄网上任何一个数字。写论文或者做实验时阈值通常定义成一张参数表方便在答辩时演示不同参数下的效果差异。参数名默认值含义调参依据SYN_THRESHOLD100时间窗口内同一源IP的SYN包上限无攻击时本机SYN包基线峰值的1.5到2倍SCAN_THRESHOLD30时间窗口内同一源IP访问的不同目的端口数上限内网合法服务数量加上合理冗余WINDOW_SEC10滑动窗口长度单位秒窗口太短误报高太长反应慢BAN_SECONDS300封禁时长单位秒攻击持续时间较短时可缩短长期对抗时调大这张表是系统的骨架逻辑代码里所有魔法数字都从这几个常量来后面不管怎么改行为先改这张表。3. 用Python把检测落到代码抓包循环、计数器与攻击判定3.1 初始化抓包线程与攻击记录表先把基础数据结构建起来。所有统计都放进字典key是源IPvalue是一个列表列表里存元素是“时间戳加目标端口”的元组。列表天然保留时间顺序清理过期数据时从头部弹出就行比维护一堆计数器变量好理解。from collections import defaultdict from scapy.all import sniff, IP, TCP import time, threading, subprocess, json # 三个核心阈值改成 2.3 节那张参数表的默认值 SYN_THRESHOLD 100 SCAN_THRESHOLD 30 WINDOW_SEC 10 BAN_SECONDS 300 # syn_record[src_ip] [(event_time, dst_port), ...] syn_record defaultdict(list) # 已经封禁的IP集合避免重复触发封禁命令 banned_ips set() # 保护 banned_ips 和 syn_record 的锁因为抓包回调和封禁线程不共用一个执行流 lock threading.Lock() def clean_expired(records, now): 把窗口外的记录全部弹出返回清理后的列表 while records and now - records[0][0] WINDOW_SEC: records.pop(0) return records逻辑说明clean_expired函数做的事就是滑动窗口淘汰每次处理新包时先调用它把超过10秒的旧记录清掉保证计数永远只看当前窗口。这里用列表头部弹出虽然时间复杂度是O(n)但每个IP每秒能产生的包数量有限实际跑起来开销可以忽略。如果有性能洁癖可以换成collections.deque它在头部弹出的效率更高但对课程设计这个量级不是必需。参数说明WINDOW_SEC直接控制检测灵敏度。设短了窗口内的样本少攻击包一多立刻触发报警误报也会变多设长了需要攒够足够多的包才触发反应延迟变大。10秒是内网实验环境的折中值公网环境建议先抓24小时流量算基线再定。3.2 单包处理逻辑从TCP标志位到攻击者画像Scapy的回调函数是整条链路的入口。每个TCP包到这里先判断是不是SYN包如果是就记录到该源IP的列表里然后调用检测函数看累计值有没有超阈值。判断SYN用tcp.flags 0x02这是TCP头里SYN标志位的二进制权重。def packet_callback(pkt): # 只处理TCP协议且带IP层的包过滤掉ARP等其他噪声 if not (pkt.haslayer(IP) and pkt.haslayer(TCP)): return ip_src pkt[IP].src tcp_layer pkt[TCP] dst_port tcp_layer.dport now time.time() # 只统计SYN包SYN Flood和端口扫描都靠SYN触发ACK建立的正常流量不参与计数 if tcp_layer.flags 0x02: with lock: syn_record[ip_src].append((now, dst_port)) syn_record[ip_src] clean_expired(syn_record[ip_src], now) # 每种攻击的判定逻辑单独拆一个函数方便答辩时逐行讲 if check_scan_attack(ip_src): block_ip(ip_src, port_scan) elif check_synflood_attack(ip_src): block_ip(ip_src, syn_flood)逻辑说明先过滤非TCP包是为了让回调函数少做无用功。判定顺序上先判断端口扫描再判断SYN Flood是因为端口扫描一定伴随大量SYN到不同端口而SYN Flood只是同一端口大量SYN先判断端口扫描会让特征更精确也防止同一次扫描行为同时触发两类攻击的封禁导致规则重复插入。3.3 攻击判定函数让计数说话判定函数是系统的核心逻辑。端口扫描统计的是“一个源IP访问了多少个不同端口”所以要把窗口内的目标端口取出来做去重再数数量。SYN Flood统计的是“窗口内源IP发来的SYN包总数”这个数字正常情况下要远小于HTTP服务或数据库服务的并发连接数。def check_scan_attack(src_ip, nowNone): now now or time.time() with lock: syn_record[src_ip] clean_expired(syn_record[src_ip], now) # 收集窗口内去重后的目标端口达到阈值即判定为扫描 dst_ports {port for _, port in syn_record[src_ip]} return len(dst_ports) SCAN_THRESHOLD def check_synflood_attack(src_ip, nowNone): now now or time.time() with lock: syn_record[src_ip] clean_expired(syn_record[src_ip], now) # 窗口内所有SYN包累加不看端口分布只看总量 return len(syn_record[src_ip]) SYN_THRESHOLD逻辑说明这两个函数的判定依据完全不同理解了它们就理解了检测原理的一半。check_scan_attack用集合推导式去重目标端口防御的是短时间探测多端口的横向侦察check_synflood_attack统计SYN包总量防御的是消耗半连接队列的纵向打击。实际攻击很多是混合型的攻击者先扫描找开放端口再对找到的端口打SYN Flood所以两个判定函数要同时保留。参数说明SYN_THRESHOLD设100意味着窗口内每秒平均最多10个SYN包超过就封禁。这个数字在真实公网服务器上过于敏感但在课程设计的局域网实验环境里完全够用。演示时可以把SYN_THRESHOLD临时调到20用一条hping3命令就能触发报警效果答辩冲击力更强。3.4 主入口和最小启动命令主函数负责启动抓包循环通过iface参数指定监听网卡用prn参数绑定回调用storeFalse避免Scapy把每个包都存在内存里导致内存爆掉。def main(): print(TCP入侵检测系统启动阈值参数如下) print(f SYN_THRESHOLD{SYN_THRESHOLD}, SCAN_THRESHOLD{SCAN_THRESHOLD}, WINDOW_SEC{WINDOW_SEC}) # storeFalse 让scapy不保留包内容只走回调逻辑省内存 sniff(prnpacket_callback, storeFalse) if __name__ __main__: main()启动命令很简单sudo python3 ids.py。sudo必须加raw socket和网卡混杂模式都需要root权限不加权限的话Scapy会在初始化时直接报错这是新手踩得最多的一个坑。要换网卡监听在sniff()里传ifaceeth0参数即可虚拟机里网卡名通常是ens33或者ens5可以用ip a命令先查看。到这一步一个能检测、能报警的检测端就落地了。但是只报警不防御在毕设演示里差一口气评委往下追问一句“检测到之后系统做了什么”如果回答只是打了行日志这个项目就停在“入侵提醒器”的水平了。防御动作放在第4章这也是标题里“联动iptables”的分量所在。4. iptables联动防御三种封禁策略与防误封设计4.1 Python里执行iptables的三种常见方式联动iptables是通过subprocess调用外部命令完成的本质是把Python算出的“攻击者IP”拼到iptables参数里然后交给系统执行。常用方式有三种封禁整个IP、限制并发连接数、按速率限流。# 方式一直接丢弃该IP的所有入站流量最干净利落 iptables -I INPUT -s 1.2.3.4 -j DROP # 方式二限制单IP对一台主机的并发连接数适合误报率较高时使用 iptables -A INPUT -p tcp --syn --dport 80 -m connlimit --connlimit-above 50 -j REJECT # 方式三按源IP限速超过10个包每秒就丢包给合法流量留活路 iptables -A INPUT -p tcp --syn -m hashlimit --hashlimit-above 10/sec --hashlimit-burst 10 --hashlimit-mode srcip -j DROP逻辑说明方式一最粗暴适合确认无误的SYN Flood源方式二限制的是并发连接数适合Web服务防CC类攻击方式三限速不会完全断掉通信适合拿不准的场景。重点提醒connlimit限制的是连接数对SYN Flood这种压根不完成握手的攻击方式效果有限它主要防的是建立大量完整连接的慢速攻击。SYN Flood的核心是半连接队列被塞满靠方式一的丢包或方式三的限速才能止血。4.2 联动封禁与自动解封Python侧的核心代码Python侧封禁动作要解决两个问题重复封禁和长时间封禁。重复封禁靠banned_ips集合做去重已经封过的IP不再发第二条iptables命令长时间封禁靠定时器自动解封避免误封IP后要手动清规则的尴尬。def block_ip(src_ip, reason, ban_secondsBAN_SECONDS): with lock: # 禁止封禁本机IP和网关这是防止断网的底线 if src_ip in (127.0.0.1, 192.168.1.1): print(f[拦截] 放弃封禁保留IP: {src_ip}) return if src_ip in banned_ips: return banned_ips.add(src_ip) cmd [iptables, -I, INPUT, -s, src_ip, -j, DROP] # checkFalse 避免iptables执行失败时抛异常capture_output 拿报错信息 result subprocess.run(cmd, capture_outputTrue, textTrue) if result.returncode ! 0: print(f[错误] iptables调用失败: {result.stderr}) with lock: banned_ips.discard(src_ip) return log {time: time.strftime(%Y-%m-%d %H:%M:%S), src_ip: src_ip, reason: reason, action: DROP} print(json.dumps(log, ensure_asciiFalse)) # 定时器到点后自动解封防止误封导致业务长时间不可用 threading.Timer(ban_seconds, unblock_ip, args(src_ip,)).start() def unblock_ip(src_ip): with lock: banned_ips.discard(src_ip) cmd [iptables, -D, INPUT, -s, src_ip, -j, DROP] subprocess.run(cmd, capture_outputTrue, textTrue) print(f[信息] 已解封 {src_ip})逻辑说明block_ip里的白名单判断是保命设计。第5章会详细讲误封网关的翻车现场代码里先把这个防线埋上。调用iptables时用capture_outputTrue把标准错误拿回来打印这样权限不足、命令拼错都能第一时间在终端看到不至于程序静默失败。threading.Timer启动一个后台线程到时间自动执行解封命令这是处理误封的后悔药。参数说明ban_seconds按场景拆开讲比较合理。做实验演示时设120秒足够看到封禁效果又不影响后续测试部署到内网服务器建议设300到600秒给运维留出人工介入的时间生产环境通常不自动解封而是把报警推给运维人工决策。参数写在block_ip函数签名里改一处全局生效。4.3 联动生效验证从nmap自测到iptables规则查看代码写完了关键是验证“检测到攻击之后到底封没封住”。这里提供一个最小自测流程一台机器跑Python检测程序另一台机器用nmap做一次半开扫描测试看结果差异。# 在攻击测试机上执行半开扫描目标是运行检测程序的机器 nmap -sS 192.168.1.100 # 在防御机上查看iptables是否已经插入规则 iptables -L INPUT -n --line-numbers # 如果规则已插入会看到类似如下输出 # 1 DROP all -- 192.168.1.88 anywhere # num为目标用iptables -D INPUT 1删除规则的序号常见疑问是“iptables配置后需要重启程序吗”答案是无需重启。iptables规则由内核netfilter即时生效Python进程只管插规则插完规则立刻开始过滤流量双方互不阻塞。自测时如果nmap第一次扫描就触发封禁第二次扫描会看到目标主机所有端口都显示filtered那就是联动生效的铁证。如果防御机上抓到了攻击包但iptables里查不到规则优先查两件事第一Python进程是否以root权限运行普通用户执行iptables命令一定失败第二程序里的block_ip是否被banned_ips集合挡住了检查日志里有没有[错误] iptables调用失败的输出。5. 实战避坑与排查从丢包到误封的五个踩坑记录5.1 抓包抓了个寂寞VMware NAT模式下看不到攻击流量现象攻击机、目标机都装在虚拟机里sniff()启动了但攻击机发SYN包目标机的日志一条都不输出。原因VMware的NAT模式会做地址转换虚拟机外部的流量根本不会以原始形态出现在虚拟网卡上。Scapy抓包是在网卡层抓的NAT模式把流量转成了宿主机和虚拟网卡之间的通信抓包程序在虚拟机内看到的只有广播和本机自己发出去的包。解决把两台虚拟机的网络模式改成桥接模式让它们像两台真实主机一样走同一段局域网。改完后用ip a确认IP地址段变成了同一网段再重新抓包验证。这是做网络安全类毕设最消耗时间的环境坑改配置五秒钟排查两小时。5.2 误封网关导致自己断网一条白名单是保命符现象检测程序跑了半天突然本机和整个内网都不通了ping网关也ping不通。原因攻击者为了掩盖身份伪造了网关IP地址来打SYN Flood。检测系统只看源IP就把它当成攻击源一条iptables -I INPUT -s 网关IP -j DROP直接把网关丢了所有依赖网关转发的流量全部中断。解决在block_ip函数里维护一个保留IP白名单本机IP、网关IP、内网DNS服务器IP默认不做封禁。更严谨一点可以做一个可配置的whitelist列表放在配置文件里部署时由管理员维护。血泪经验这条防线必须在联网前加上否则演示翻车现场就是这个场景。5.3 connlimit能防攻防防不了SYN Flood现象按网上的教程加了-m connlimit --connlimit-above 50 -j REJECT测试SYN Flood时发现目标机该卡还是卡半连接队列照样被打满。原因connlimit限制的是并发连接数它需要连接先建立起来才会计数。SYN Flood攻击的特点是攻击包永远停留在SYN阶段三次握手根本走不完连接从未建立connlimit对它们完全无效。这个命令真正适合的场景是限制单IP并发建立的完整连接数比如防Web爬虫拉爆连接池。解决SYN Flood必须用丢包或限速来防。-j DROP直接扔掉攻击包不进协议栈-m hashlimit限制了单源IP的SYN包速率。理解这个区别答辩被追问时也站得住脚。5.4 iptables规则无限增长封禁和查重必须成对出现现象系统跑了一小时iptables -L INPUT -n输出几百条规则全部是同一个攻击IP的DROP记录。原因检测逻辑在每次收到攻击包时都调用block_ip没有判断这个IP是否已经在封禁名单里。攻击流量持续进来封禁命令就持续插入规则表越积越长最终影响内核包过滤效率。解决代码里用banned_ips集合先做一次去重判断如果IP已经在集合里就跳过。同时配合threading.Timer自动解封封禁到期的IP从集合里移除后续再有攻击包进来时可以重新封禁。这套“查重、封禁、定时解封”的闭环在流量攻击场景下是必须的能抵挡高频攻击而不拖垮系统。5.5 Scapy抓包偶尔漏包高流量下判定不准确现象用小流量hping3测试时一切正常一旦用nmap -sS -p 1-65535全端口扫描检测日志里的SYN包数量明显少于实际发送量判定时灵时不灵。原因Scapy是用户态抓包库数据包从网卡到内核再到Python进程有一整条链路要跑回调函数处理速度跟不上时内核缓冲区满了就开始丢包。全端口扫描产生的SYN包速度远超Scapy的处理能力丢包就是必然结果。解决抓包回调只做轻量操作把所有数据先存进队列后台单独开一个线程做检测判定。改完线程模型后Scapy的丢包率能明显下降虽然依旧不是线速抓包但应对课程设计和毕设级别的流量绰绰有余。如果对性能有更高要求可以把抓包层换成libpcap的Python绑定不过那个改动会额外增加一轮解析成本非必要不动。6. 把系统做得更像工程hping3自测闭环和日志可视化检测和封禁跑通之后离“能演示的项目”还差一道工序就是自测。每次改完阈值参数先用自测脚本验证一遍再上真实网段不然误报误封的压力会全堆在某个倒霉的深夜。自测工具用hping3和nmap端口扫描的组合就够了# 模拟SYN Flood单条命令触发封禁逻辑 hping3 -S -p 80 -i u1000 192.168.1.100 # 模拟端口扫描验证SCAN_THRESHOLD判定 nmap -sS -p 1-300 192.168.1.100跑完看两件事程序终端是否打印了json格式的攻击日志iptables -L INPUT -n里是否出现了对应源IP的DROP规则。两件都正常联动链路就是通的。工程化方面可以加一层日志可视化把block_ip里的json.dumps输出写到独立的日志文件再用Flask写一个十几行的展示页面读日志按时间轴展示攻击IP和封禁行为。这块不需要做得多复杂能画出一张攻击时间线表答辩的观感就会拉高一个档次。我自己的习惯是改任何一个检测阈值后先跑十分钟自测再放上真实网段。这套流程保证系统不会因为一个粗糙的参数设置把整个内网网关封了。希望帮到你也祝这个毕设拿个好成绩。本文还有配套的精品资源点击获取
返回列表