简介这是一套面向高校计算机网络、信息安全专业学生及中小型网络运维人员的Python TCP入侵检测系统源码可作为毕业设计、课程设计或项目开发的技术参考。系统聚焦TCP层面的安全监测与主动防御通过分析连接请求的时间序列频率、TCP头部标志位组合SYN、FIN及NULL包分布以及非监听端口连接占比识别端口扫描与DDoS攻击行为并借助scapy抓包解析、python-iptables联动防火墙、MySQLdb存储检测日志形成检测到响应的闭环。资源包共10个文件以5个py源码文件为核心辅以3个zbak备份、1个zip与1个md说明文档整体约10KB结构清晰、模块耦合度低。目前已有41人学习。读者可从中获取完整的检测逻辑实现、防火墙联动策略与数据库日志设计思路适合作为网络安全实践教学案例或主动防护体系的扩展基础。1. 从一台被扫穿的测试机说起TCP 入侵检测到底在防什么去年帮朋友看一台暴露在公网的测试服务器日志里全是同一段模式某个 IP 在 3 秒内连了 2000 多个端口紧接着就是一波半连接请求把 accept 队列打满业务进程直接卡死。事后复盘问题不在于没装防火墙而在于没人盯着「连接行为」本身。基于 Python 的 TCP 入侵检测系统干的就是这件事在传输层抓包或读连接状态识别端口扫描和 DoS 攻击这两类最典型的异常再通过 iptables 把攻击源拦在门外。它适合做毕业设计、课程设计也适合中小规模服务做一层轻量级行为监控。核心链路是「抓包 → 特征判定 → 告警 → 联动封禁」Python 负责前两步和调度iptables 负责最后一步执行。下面按这条链路拆开讲重点放在能跑起来的代码和参数上。2. 抓包与连接状态采集Python 侧怎么拿到可信数据2.1 两种采集路线的选型理由做 TCP 入侵检测第一步是拿到数据。常见做法有两类一是用 scapy 直接抓原始包二是读/proc/net/tcp或调用ss拿连接状态。两者不是替代关系而是互补。scapy 的优势是能看到 TCP 标志位SYN、ACK、RST、FIN 一目了然判断端口扫描和 SYN Flood 靠的就是这些标志位的组合。缺点是 Python 层抓包在高流量下会丢包单进程处理超过几万 pps 就开始吃力。读/proc/net/tcp的优势是零抓包开销内核已经维护好了连接状态适合做连接数统计和长连接异常检测缺点是看不到单个包的时序判断不了「半连接」这种瞬时行为。我一般会两条路都留着scapy 负责短时间窗口内的包级特征/proc/net/tcp负责连接数基线。毕业设计里如果只选一条优先 scapy因为端口扫描和 SYN Flood 的判定逻辑更直观答辩时也好讲。2.2 用 scapy 抓 TCP 包的最小可跑代码from scapy.all import sniff, TCP, IP from collections import defaultdict import time # 记录每个源IP在时间窗口内访问过的目标端口 port_scan_tracker defaultdict(set) # 记录每个源IP的SYN包时间戳 syn_tracker defaultdict(list) WINDOW 5 # 时间窗口单位秒 PORT_THRESHOLD 20 # 窗口内访问不同端口数超过此值判定为扫描 SYN_THRESHOLD 100 # 窗口内SYN包数超过此值判定为SYN Flood def process_packet(pkt): if not (pkt.haslayer(IP) and pkt.haslayer(TCP)): return src_ip pkt[IP].src dst_port pkt[TCP].dport flags pkt[TCP].flags now time.time() # 端口扫描特征同一源IP在窗口内访问大量不同目标端口 port_scan_tracker[src_ip].add(dst_port) if len(port_scan_tracker[src_ip]) PORT_THRESHOLD: print(f[扫描告警] {src_ip} 在 {WINDOW}s 内访问了 {len(port_scan_tracker[src_ip])} 个端口) port_scan_tracker[src_ip].clear() # SYN Flood特征窗口内大量SYN包且没有对应的ACK完成握手 if flags S: syn_tracker[src_ip].append(now) syn_tracker[src_ip] [t for t in syn_tracker[src_ip] if now - t WINDOW] if len(syn_tracker[src_ip]) SYN_THRESHOLD: print(f[SYN Flood告警] {src_ip} 在 {WINDOW}s 内发送了 {len(syn_tracker[src_ip])} 个SYN) syn_tracker[src_ip].clear() sniff(filtertcp, prnprocess_packet, store0)这段代码的逻辑很直白port_scan_tracker用集合记录每个源 IP 在窗口内碰过的端口集合大小超过阈值就说明它在扫。syn_tracker记录 SYN 包时间戳只保留窗口内的数量超限就判定为 SYN Flood。sniff的filtertcp是 BPF 语法把非 TCP 包直接在内核层过滤掉减少 Python 层压力。参数上WINDOW设 5 秒是经验值太短会漏掉慢速扫描太长会把正常业务的端口访问误判。PORT_THRESHOLD设 20 适合中小型服务如果业务本身会连很多端口要往上调。SYN_THRESHOLD设 100 是保守值真实环境里 SYN Flood 往往是每秒上千个但设太低会误伤 NAT 后面的正常用户。提示scapy 需要 root 权限或CAP_NET_RAW能力才能抓包Windows 上还要装 Npcap。如果跑在容器里记得加--cap-addNET_RAW。2.3 连接状态采集的补充脚本import subprocess def get_tcp_connections(): # 用ss命令拿当前TCP连接状态-t只显示TCP-a显示所有-n不解析域名 result subprocess.run( [ss, -tan], capture_outputTrue, textTrue ) lines result.stdout.strip().split(\n)[1:] conn_count {} for line in lines: parts line.split() if len(parts) 5: continue state parts[0] peer parts[4] # 统计每个对端IP的连接数 ip peer.rsplit(:, 1)[0] conn_count[ip] conn_count.get(ip, 0) 1 return conn_count if __name__ __main__: counts get_tcp_connections() for ip, cnt in sorted(counts.items(), keylambda x: -x[1])[:10]: print(f{ip}: {cnt} 条连接)这个脚本用ss -tan拿全量 TCP 连接按对端 IP 聚合。它的价值在于发现「大量 ESTABLISHED 连接来自同一 IP」这种慢速 DoSscapy 抓包不一定能及时反映。ss比netstat快在连接数上万时差距明显。输出取前 10 个连接数最多的 IP方便快速定位异常源。2.4 采集层的关键参数与边界抓包和连接状态采集都有性能边界。scapy 单进程在普通虚拟机上大概能处理 1 到 2 万 pps超过这个量级要么上多进程要么改用 C 扩展或 PF_RING。/proc/net/tcp读取本身很快但解析 10 万条连接需要几百毫秒不适合秒级高频调用。另一个边界是权限。抓包要 root读/proc/net/tcp普通用户只能看到自己的连接要拿全量也得 root 或CAP_NET_ADMIN。做毕业设计时如果部署在云主机上注意安全组和系统防火墙不要和 iptables 规则冲突否则会出现「规则写了但不生效」的玄学问题。3. 端口扫描与 DoS 攻击的判定逻辑阈值怎么定才不误伤3.1 端口扫描的三种模式与识别差异端口扫描不是只有一种。最常见的三种TCP connect 扫描、SYN 半开扫描、FIN/XMAS 隐蔽扫描。它们的包特征完全不同判定逻辑也要分开。TCP connect 扫描会完成三次握手日志里能看到大量 ESTABLISHED 或 TIME_WAIT 连接目标端口分散。SYN 半开扫描只发 SYN收到 SYN-ACK 后直接发 RST 断开不会完成握手特征是大量 SYN 包但没有对应的 ACK。FIN 扫描发的是 FIN 包绕过一些只检测 SYN 的防火墙。我一般用「窗口内不同目标端口数」作为主判据因为它对三种扫描都有效。辅助判据是「SYN 与 ACK 的比例」正常业务 SYN 和 ACK 数量接近扫描场景下 SYN 远多于 ACK。3.2 阈值设定的实操方法阈值不能拍脑袋。我的做法是先跑一段基线在业务正常时段采集 10 分钟数据统计每个源 IP 在 5 秒窗口内访问不同端口数的分布取 P99 作为阈值下限再往上留 20% 余量。import numpy as np # 假设baseline是正常时段采集的每个元素是某IP在5s窗口内的端口数 baseline [3, 5, 2, 8, 4, 6, 3, 7, 5, 4, 9, 3, 5, 6, 4] p99 np.percentile(baseline, 99) threshold int(p99 * 1.2) print(fP99{p99:.1f}, 建议阈值{threshold})这段代码用 numpy 算 P99 再乘 1.2。如果基线里最大才 9阈值大概在 11 到 12比拍脑袋设 20 更贴合实际。DoS 的 SYN 阈值同理先统计正常时段每秒 SYN 数取峰值乘 2 到 3 倍。注意阈值是动态的。业务上线新功能、做压测、搞活动时正常流量模式会变阈值要跟着调。我见过有人设完阈值就不管结果大促时把正常用户全封了血泪经验。3.3 把判定结果写成结构化告警告警不能只 print要结构化方便后续联动和审计。import json import datetime def build_alert(alert_type, src_ip, detail, severityhigh): alert { timestamp: datetime.datetime.now().isoformat(), type: alert_type, src_ip: src_ip, detail: detail, severity: severity } return json.dumps(alert, ensure_asciiFalse) # 使用示例 alert_json build_alert( port_scan, 192.168.1.100, {window: 5, port_count: 45, threshold: 20} ) print(alert_json)build_alert把告警类型、源 IP、细节和严重级别打包成 JSON。ensure_asciiFalse保证中文正常显示。这个结构可以直接写日志文件也可以发给消息队列或 webhook。严重级别用来区分「扫描」和「DoS」后者应该触发更激进的封禁策略。3.4 误报与漏报的权衡入侵检测永远在误报和漏报之间走钢丝。阈值调低误报多正常用户被误封阈值调高漏报多攻击者慢速扫描能溜过去。我的经验是端口扫描宁可误报一点因为封禁一个扫描源的成本很低解封也快。DoS 判定要保守因为 SYN Flood 的阈值设太低NAT 后面几百个正常用户可能被一起封掉。折中方案是分级第一次触发只告警不封禁同一 IP 在 10 分钟内触发 3 次才执行封禁。这样给正常用户留了缓冲对攻击者也有足够威慑。4. iptables 联动防御从告警到封禁的自动化链路4.1 为什么选 iptables 而不是其他方案联动防御的可选项不少iptables、nftables、fail2ban、云厂商安全组 API。选 iptables 的理由很实际Linux 默认就有不需要额外装服务规则生效在 netfilter 层性能损耗小而且毕业设计里讲起来链路清晰。nftables 是更新一代语法更统一但很多老系统默认还是 iptables教学环境里 iptables 资料更多。fail2ban 本质也是调 iptables自己写一遍能讲清楚原理答辩时更有说服力。4.2 用 Python 调用 iptables 封禁 IPimport subprocess def block_ip(ip, reasonids_alert): # 先检查规则是否已存在避免重复添加 check subprocess.run( [iptables, -C, INPUT, -s, ip, -j, DROP], capture_outputTrue ) if check.returncode 0: print(f{ip} 已在封禁列表中) return False # 添加DROP规则-I插入到链首保证优先生效 result subprocess.run( [iptables, -I, INPUT, -s, ip, -j, DROP, -m, comment, --comment, reason], capture_outputTrue, textTrue ) if result.returncode 0: print(f已封禁 {ip}) return True else: print(f封禁失败: {result.stderr}) return False def unblock_ip(ip): result subprocess.run( [iptables, -D, INPUT, -s, ip, -j, DROP], capture_outputTrue, textTrue ) return result.returncode 0block_ip先用-C检查规则是否存在避免重复插入导致规则链膨胀。-I INPUT把规则插到链首保证优先匹配。-m comment --comment加备注方便后续审计和批量清理。unblock_ip用-D删除规则。参数上-s指定源 IP-j DROP直接丢包。如果只想限速不想全封可以换成-j ACCEPT配合-m limit但 DoS 场景下 DROP 更干脆。4.3 封禁策略永久、临时还是分级直接永久封禁风险很大。攻击者可能用动态 IP封了也没用正常用户被误封永久规则会一直影响他。我一般用分级策略触发次数时间窗口动作封禁时长1 次10 分钟仅告警无2 次10 分钟临时封禁5 分钟3 次及以上10 分钟长期封禁1 小时临时封禁用at命令或后台定时任务解封。下面是一个简单的定时解封实现import threading def block_ip_with_timeout(ip, timeout300): if block_ip(ip): timer threading.Timer(timeout, unblock_ip, args[ip]) timer.daemon True timer.start() print(f{ip} 将在 {timeout}s 后自动解封)threading.Timer在后台线程里等 timeout 秒后调unblock_ip。daemonTrue保证主进程退出时定时器不会阻塞。这个方案适合单机如果要多机联动得把封禁状态写到 Redis 或数据库里统一管理。4.4 规则持久化与开机恢复iptables 规则默认重启就丢。生产环境要用iptables-persistent或netfilter-persistent保存。但入侵检测系统动态添加的规则不建议直接持久化否则重启后一堆临时封禁变成永久的。我的做法是只持久化基础规则比如放行 SSH、HTTP动态封禁规则存在一个 JSON 文件里系统启动时由 IDS 进程读取并重新下发。这样封禁状态可控也方便审计。import json import os BLOCKLIST_FILE /var/lib/ids/blocklist.json def save_blocklist(blocked_ips): os.makedirs(os.path.dirname(BLOCKLIST_FILE), exist_okTrue) with open(BLOCKLIST_FILE, w) as f: json.dump(blocked_ips, f) def load_blocklist(): if not os.path.exists(BLOCKLIST_FILE): return {} with open(BLOCKLIST_FILE) as f: return json.load(f)save_blocklist把当前封禁的 IP 和封禁时间写文件load_blocklist启动时读取。配合定时任务清理过期条目就能做到重启后状态可恢复。5. 避坑与排查那些让系统「看起来在跑但没生效」的问题5.1 抓不到包但进程没报错现象scapy 的sniff跑着日志里一条包都没有进程也不退出。原因最常见的是权限不够。普通用户抓包scapy 可能不报错但收不到任何包。其次是网卡选错了容器里默认抓eth0但实际流量走的是veth或br-开头的网桥接口。解决用ip a确认流量走哪个接口sniff里显式指定ifaceeth0。权限问题用sudo或给 Python 加CAP_NET_RAW。容器里加--cap-addNET_RAW --cap-addNET_ADMIN。5.2 iptables 规则加了但不生效现象iptables -L能看到 DROP 规则但攻击 IP 还是能连进来。原因规则链顺序不对。如果 INPUT 链前面有 ACCEPT 规则匹配了该流量后面的 DROP 永远不会执行。另一个可能是流量根本没走 INPUT 链比如被 Docker 的 FORWARD 链或 nat 表规则拦截了。解决用iptables -L INPUT -n --line-numbers看规则顺序把 DROP 插到最前面。Docker 环境要注意DOCKER-USER链容器流量默认不走 INPUT。排查时用iptables -t nat -L -n看 nat 表有没有冲突规则。5.3 阈值设太低正常用户被误封现象告警日志里出现大量正常业务 IP用户反馈连不上服务。原因阈值没有基于实际基线设定或者业务模式变了没同步调整。NAT 环境下多个用户共享一个出口 IP端口访问数天然偏高。解决先关掉自动封禁只告警跑一天看误报率。对 NAT 出口 IP 加白名单或者把阈值按 IP 段区分。我一般会留一个WHITELIST列表里面放网关、监控、办公网出口。5.4 高流量下 Python 进程 CPU 打满现象流量一上来IDS 进程 CPU 100%抓包丢包严重告警延迟。原因scapy 在 Python 层逐包处理GIL 限制了多线程并行单进程吞吐有上限。解决用 BPF filter 在内核层先过滤只抓关心的包。把判定逻辑拆到多进程每个进程抓不同端口范围。极端情况下改用PF_RING或AF_PACKET的 mmap 模式。毕业设计里如果流量不大优化 filter 就够了。5.5 封禁后自己也被封了现象调试时用 SSH 连服务器触发规则后 SSH 断开再也连不上。原因测试用的源 IP 和 SSH 登录 IP 是同一个规则把管理流量也封了。解决调试前先把自己 IP 加白名单或者用带外管理云控制台的 VNC。iptables 规则里加一条-s 管理IP -j ACCEPT放在最前面。这个坑我踩过不止一次后悔药就是提前留好带外通道。6. 进阶把检测规则做成可配置的 YAML 策略6.1 为什么要把规则外置硬编码在 Python 里的阈值和逻辑改一次要重启进程毕业设计答辩时也不好展示「可扩展性」。把规则抽成 YAML检测引擎读配置执行改阈值不用动代码还能按不同业务场景加载不同策略文件。# rules.yaml port_scan: window: 5 port_threshold: 20 severity: high action: block syn_flood: window: 5 syn_threshold: 100 severity: critical action: block whitelist: - 192.168.1.1 - 10.0.0.0/8 block_policy: first_trigger: alert second_trigger: block_300 third_trigger: block_36006.2 加载配置并驱动检测引擎import yaml def load_rules(pathrules.yaml): with open(path) as f: return yaml.safe_load(f) class IDSEngine: def __init__(self, rules): self.rules rules self.whitelist set(rules.get(whitelist, [])) self.trigger_count {} def is_whitelisted(self, ip): # 简化版实际要支持CIDR匹配 return ip in self.whitelist def handle_alert(self, ip, alert_type): if self.is_whitelisted(ip): return cnt self.trigger_count.get(ip, 0) 1 self.trigger_count[ip] cnt policy self.rules[block_policy] if cnt 1: print(f[告警] {ip} 触发 {alert_type}) elif cnt 2: block_ip_with_timeout(ip, 300) else: block_ip_with_timeout(ip, 3600)load_rules用yaml.safe_load读配置避免执行任意代码。IDSEngine把白名单和触发计数封装起来handle_alert按触发次数走分级策略。白名单匹配这里简化成精确匹配实际要支持 CIDR可以用ipaddress模块。6.3 验证规则是否按预期工作改完规则别直接上生产先用测试流量验证。我一般用hping3或nmap在另一台机器上模拟扫描看 IDS 是否告警、iptables 是否封禁、解封是否按时执行。# 模拟SYN扫描-S发SYN-p指定端口范围-i u1000每1000微秒发一个 sudo hping3 -S -p 1-100 -i u1000 192.168.1.50 # 查看iptables规则是否命中 sudo iptables -L INPUT -n -v --line-numbers # 查看封禁日志 tail -f /var/log/ids/alert.log验证时重点看三个点告警延迟是否在可接受范围一般 1 到 2 秒、封禁规则是否插到了正确位置、解封后规则是否真的被删除。如果用了threading.Timer注意进程重启后定时器会丢解封逻辑要配合持久化文件做补偿。6.4 一个我常犯的错误早期做这个系统时我把所有逻辑塞在一个while True循环里抓包、判定、封禁串行执行。结果一次 iptables 调用卡住整个检测链路就停了攻击者趁机扫了个遍。后来改成抓包和判定在一个线程封禁操作丢到队列里由单独线程消费即使 iptables 响应慢也不会阻塞检测。这个习惯一直保留到现在检测链路和执行链路必须解耦中间用队列缓冲。队列长度要设上限满了就丢最老的告警宁可漏封也不能让检测停摆。希望帮到你。本文还有配套的精品资源点击获取