简介本资源是一套基于Python实现的轻量级TCP入侵检测系统源码面向计算机网络、信息安全等专业的本科生课程设计与毕业设计实践聚焦端口扫描与DoS攻击的实时识别及iptables联动防御。系统通过分析TCP连接频率、标志位组合异常如SYN洪泛、NULL扫描及非监听端口访问占比三类特征完成威胁判定技术栈涵盖scapy抓包解析、python-iptables策略控制与MySQLdb日志存储代码模块解耦清晰具备教学演示与中小网络环境部署扩展能力。压缩包共10个文件含5个核心Python脚本如Main.py主控、Analysis.py特征分析、Flitter.py流量过滤、3个备份文件.zbak、1个README说明文档及1个嵌套ZIP整体仅10KB便于快速导入与调试。目前已有41人学习下载提供完整可运行工程结构、关键算法注释、防火墙联动逻辑实现及MySQL日志存储方案是理解网络层IDS原理与落地实践的优质参考案例。1. 这不是“写个脚本抓包就叫入侵检测”一个能真正在Linux服务器上跑起来、自动封IP、扛住SYN Flood的TCP层防御系统你手头有个Python写的“入侵检测”脚本它能打印出“发现端口扫描”但没人敢把它放到生产环境的Web服务器上——因为一跑就吃光CPU封错IP导致自己业务断连或者面对真实SYN Flood时像纸糊的一样被冲垮。这篇讲的是毕业设计里极少见、但企业侧真在用的落地路径用纯Python解析原始TCP数据包不依赖Scapy发包只做被动监听结合连接状态机建模识别慢速扫描与半开连接异常再通过iptables -I INPUT -s xxx.xxx.xxx.xxx -j DROP 实现毫秒级联动封禁。它不碰应用层协议HTTP/FTP专注TCP三次握手、RST/FIN序列、重传超时等底层行为不靠机器学习黑匣子用滑动窗口连接存活时间并发连接斜率三个硬指标做判定所有代码可单文件部署无需数据库、不拉Docker、不改内核参数。适合网络工程课设、安全方向毕设、中小IDC运维自建轻量防护也适合作为理解Linux netfilter机制的实操入口。2. 从raw socket到连接状态机为什么必须绕过Scapy、自己解析TCP头部2.1 为什么不用Scapy做实时检测血泪经验告诉你瓶颈在哪Scapy功能强大但它的包解析是Python层逐字节解码对象封装对每秒上万TCP SYN包的场景CPU占用直接飙到90%以上。我曾用Scapy监听eth0在3000pps的SYN Flood下解析延迟从2ms涨到800ms漏检率超40%。真正能扛住压力的做法是绕过Scapy用socket.AF_PACKET socket.SOCK_RAW直通网卡驱动用ctypes结构体映射IP/TCP头部零拷贝提取源IP、目的端口、标志位、序列号。这不是炫技——当你的检测逻辑必须在50μs内完成一次判断时少一次Python对象创建就少一次GC停顿。import socket import ctypes import struct class IPHeader(ctypes.Structure): _fields_ [ (ihl, ctypes.c_ubyte, 4), (version, ctypes.c_ubyte, 4), (tos, ctypes.c_ubyte), (len, ctypes.c_ushort), (id, ctypes.c_ushort), (offset, ctypes.c_ushort), (ttl, ctypes.c_ubyte), (protocol, ctypes.c_ubyte), (checksum, ctypes.c_ushort), (src, ctypes.c_uint32), (dst, ctypes.c_uint32) ] class TCPHeader(ctypes.Structure): _fields_ [ (src_port, ctypes.c_ushort), (dst_port, ctypes.c_ushort), (seq, ctypes.c_uint32), (ack, ctypes.c_uint32), (offset, ctypes.c_ubyte, 4), (reserved, ctypes.c_ubyte, 3), (flags, ctypes.c_ubyte, 1), # URG, ACK, PSH, RST, SYN, FIN (window, ctypes.c_ushort), (checksum, ctypes.c_ushort), (urgent_ptr, ctypes.c_ushort) ] def parse_tcp_packet(raw_data: bytes) - dict: if len(raw_data) 34: # 最小IP(20)TCP(20)长度 return {} ip_hdr IPHeader.from_buffer_copy(raw_data[:20]) if ip_hdr.protocol ! 6: # 6 TCP return {} tcp_hdr TCPHeader.from_buffer_copy(raw_data[20:40]) src_ip socket.inet_ntoa(struct.pack(!I, ip_hdr.src)) dst_port socket.ntohs(tcp_hdr.dst_port) flags tcp_hdr.flags return { src_ip: src_ip, dst_port: dst_port, syn: bool(flags 0x02), # SYN flag bit ack: bool(flags 0x10), # ACK flag bit rst: bool(flags 0x04), # RST flag bit fin: bool(flags 0x01), # FIN flag bit seq: socket.ntohl(tcp_hdr.seq), ack_seq: socket.ntohl(tcp_hdr.ack) }提示ctypes.Structure比struct.unpack()快3倍以上且避免了字节序转换错误。关键点在于from_buffer_copy()直接内存映射不触发Python字符串解码。ip_hdr.protocol ! 6是第一道过滤丢弃所有非TCP包省下70% CPU。2.2 端口扫描检测不是数“连了几个端口”而是看“连接怎么死的”常见误区是统计某IP在1分钟内连了20个不同端口就判为扫描。但真实攻击者会用nmap -sS -T2慢速SYN扫描间隔2秒发一个SYN1小时内扫完65535个端口——这种根本不会触发阈值。真正有效的检测逻辑是追踪每个IP的TCP连接生命周期识别“SYN_SENT但无ACK返回”、“SYNACK后无ACK确认”、“RST风暴”三类异常模式。我们维护一个字典conn_state {ip: {syn_sent: set(), syn_ack_recv: set(), rst_count: 0, last_seen: time.time()}}每收到一个包更新状态收到SYN→ 加入syn_sent收到SYNACK→ 从syn_sent移除加入syn_ack_recv收到ACK且src_ip在syn_ack_recv中→ 视为连接建立成功清空该IP状态收到RST→rst_count 1每30秒扫描一次字典若len(syn_sent) 3 and time.time() - last_seen 5→ 判定为慢速扫描若rst_count 10 in 10s→ 判定为RST泛洪from collections import defaultdict, deque import time class PortScanDetector: def __init__(self, scan_window30, rst_threshold10, rst_window10): self.conn_state defaultdict(lambda: { syn_sent: set(), syn_ack_recv: set(), rst_times: deque(), # 存储RST时间戳 last_seen: time.time() }) self.scan_window scan_window self.rst_threshold rst_threshold self.rst_window rst_window def update(self, pkt: dict): ip pkt[src_ip] now time.time() self.conn_state[ip][last_seen] now if pkt[syn] and not pkt[ack]: self.conn_state[ip][syn_sent].add(pkt[dst_port]) elif pkt[syn] and pkt[ack]: self.conn_state[ip][syn_sent].discard(pkt[dst_port]) self.conn_state[ip][syn_ack_recv].add(pkt[dst_port]) elif pkt[rst]: self.conn_state[ip][rst_times].append(now) # 清理过期RST时间戳 while (self.conn_state[ip][rst_times] and now - self.conn_state[ip][rst_times][0] self.rst_window): self.conn_state[ip][rst_times].popleft() def check_scan(self, ip: str) - str: state self.conn_state[ip] now time.time() # 慢速SYN扫描SYN发出后5秒内无响应且积压3个 if (len(state[syn_sent]) 3 and now - state[last_seen] 5): return slow_syn_scan # RST泛洪10秒内RST超过10次 if len(state[rst_times]) self.rst_threshold: return rst_flood # 快速全端口扫描1秒内SYN请求端口数50针对nmap -sT if (len(state[syn_sent]) 50 and now - state[last_seen] 1): return fast_port_scan return 参数说明scan_window不是检测窗口而是状态清理周期防止内存泄漏rst_threshold和rst_window需根据业务调整——高并发Web服务RST本就多建议先抓包统计基线值再设阈值。check_scan()返回字符串而非布尔值方便后续扩展日志分类。2.3 DoS攻击检测盯紧“半开连接”的数量斜率而不是绝对值SYN Flood的本质是耗尽服务端SYN Queue即net.ipv4.tcp_max_syn_backlog。传统做法是监控ss -s | grep SYNs但这是采样值滞后2秒以上。我们的方案是每200ms统计一次当前处于SYN_RECV状态的连接数计算其10秒滑动窗口内的变化斜率Δcount/Δt。正常业务斜率0.5而SYN Flood攻击启动瞬间斜率可达50。关键在于——不依赖/proc/net/snmp直接从raw socket包流中实时计数import threading import time class DosDetector: def __init__(self, slope_window10, sample_interval0.2): self.syn_recv_counts deque(maxlen50) # 存10秒数据50*0.210 self.sample_interval sample_interval self.slope_window slope_window self.running False self.lock threading.Lock() def start_monitoring(self): self.running True t threading.Thread(targetself._monitor_loop) t.daemon True t.start() def _monitor_loop(self): while self.running: # 此处应调用parse_tcp_packet()获取当前包流中的SYN_RECV数 # 实际部署时此处接入第2.1节的包解析循环 current_count self._count_syn_recv_in_stream() # 伪代码见下文 with self.lock: self.syn_recv_counts.append((time.time(), current_count)) time.sleep(self.sample_interval) def _count_syn_recv_in_stream(self) - int: # 真实实现遍历conn_state字典统计满足以下条件的IP数 # - 在syn_sent中有端口记录 # - 且last_seen在2秒内排除已超时的旧SYN count 0 now time.time() for ip, state in list(self.conn_state.items()): if (state[syn_sent] and now - state[last_seen] 2): count len(state[syn_sent]) return count def get_slope(self) - float: with self.lock: if len(self.syn_recv_counts) 2: return 0.0 # 取最近10秒数据 window [(t, c) for t, c in self.syn_recv_counts if time.time() - t self.slope_window] if len(window) 2: return 0.0 t0, c0 window[0] t1, c1 window[-1] return (c1 - c0) / (t1 - t0) if t1 t0 else 0.0 def is_dos_active(self, slope_threshold20.0) - bool: slope self.get_slope() return slope slope_threshold注意_count_syn_recv_in_stream()不是查ss命令而是复用第2.2节的conn_state字典——这保证了检测与解析完全同步无采样延迟。slope_threshold20.0是经验值需在测试环境用hping3 -S -p 80 -i u10000 target_ip校准u10000100pps此时斜率约15~18u5000200pps时斜率达35。3. iptables联动不是简单调system()而是用nfqueue做原子级封禁3.1 为什么system(iptables -I INPUT -s %s -j DROP)是定时炸弹直接调os.system()插入iptables规则有三大致命问题竞态条件两个线程同时封同一IP生成重复规则iptables -L --line-numbers能看到第3条和第7条都是-s 1.1.1.1规则爆炸每秒封10个IP1小时就是36000条规则iptables匹配变O(n)复杂度整个INPUT链卡死无法撤回没记录规则编号iptables -D INPUT 1可能删错行。正确做法是用netfilterqueue绑定NFQUEUE让内核把匹配的包送到用户态由Python决定是否丢弃——这本质是“在数据包进入INPUT链前做决策”比iptables规则更前置、更可控。# 先配置iptables跳转到NFQUEUE sudo iptables -I INPUT -p tcp --dport 22 -m state --state NEW -j NFQUEUE --queue-num 0 sudo iptables -I INPUT -p tcp --dport 80 -m state --state NEW -j NFQUEUE --queue-num 0from netfilterqueue import NetfilterQueue import os class IptablesDefender: def __init__(self, queue_num0): self.queue_num queue_num self.blocked_ips set() self.nfqueue None def start(self): self.nfqueue NetfilterQueue() self.nfqueue.bind(self.queue_num, self.callback) self.nfqueue.run() def callback(self, packet): # 解析IP头部获取源IP复用2.1节的parse_tcp_packet try: pkt_dict parse_tcp_packet(bytes(packet.get_payload())) if not pkt_dict or src_ip not in pkt_dict: packet.accept() return ip pkt_dict[src_ip] # 检查是否已在封禁列表 if ip in self.blocked_ips: packet.drop() # 直接丢弃不进iptables return except Exception as e: pass # 解析失败放行 finally: packet.accept() # 默认放行 def block_ip(self, ip: str): if ip in self.blocked_ips: return self.blocked_ips.add(ip) # 同时写入iptables作持久化重启后仍生效 os.system(fiptables -I INPUT -s {ip} -j DROP 2/dev/null) print(f[DEFEND] Blocked {ip}) def unblock_ip(self, ip: str): if ip not in self.blocked_ips: return self.blocked_ips.remove(ip) # 删除iptables规则需先查编号 result os.popen(fiptables -L INPUT --line-numbers | grep {ip} ).read() if result: line_num result.split()[0] os.system(fiptables -D INPUT {line_num} 2/dev/null)关键逻辑packet.drop()是内核态丢包毫秒级iptables -I是用户态规则写入用于持久化和兜底。self.blocked_ips是内存白名单避免重复操作。os.system()调用带2/dev/null屏蔽错误因规则可能已存在。3.2 封禁策略分级临时封禁 vs 永久封禁用时间戳做自动轮转粗暴封禁所有可疑IP会导致误伤。我们实现三级策略Level 1临时封禁检测到慢速扫描封禁300秒到期自动解封Level 2加强封禁检测到SYN Flood封禁3600秒并写入/etc/iptables.rules持久化Level 3永久封禁同一IP24小时内触发Level 2≥3次写入/etc/iptables.blacklist重启不消失。import json import atexit class SmartBlocker: def __init__(self, temp_ttl300, hard_ttl3600): self.temp_blocks {} # {ip: expire_time} self.hard_blocks set() # 持久化封禁IP self.blacklist set() # 永久黑名单 self.temp_ttl temp_ttl self.hard_ttl hard_ttl self.load_persistence() def load_persistence(self): try: with open(/etc/iptables.blacklist, r) as f: self.blacklist set(line.strip() for line in f if line.strip()) except FileNotFoundError: pass def block(self, ip: str, level1): now time.time() if ip in self.blacklist: return blacklisted if level 1: self.temp_blocks[ip] now self.temp_ttl elif level 2: self.hard_blocks.add(ip) self._persist_hard_block(ip) elif level 3: self.blacklist.add(ip) self._persist_blacklist(ip) return blocked def _persist_hard_block(self, ip: str): os.system(fiptables -I INPUT -s {ip} -j DROP 2/dev/null) with open(/etc/iptables.rules, a) as f: f.write(fiptables -I INPUT -s {ip} -j DROP\n) def _persist_blacklist(self, ip: str): with open(/etc/iptables.blacklist, a) as f: f.write(f{ip}\n) def cleanup_expired(self): now time.time() expired [ip for ip, exp in self.temp_blocks.items() if exp now] for ip in expired: del self.temp_blocks[ip] def is_blocked(self, ip: str) - bool: if ip in self.blacklist: return True if ip in self.hard_blocks: return True if ip in self.temp_blocks and self.temp_blocks[ip] time.time(): return True return False # 程序退出时保存状态 def save_on_exit(blocker: SmartBlocker): with open(/var/run/tcp_defender.state, w) as f: json.dump({ temp_blocks: blocker.temp_blocks, hard_blocks: list(blocker.hard_blocks), blacklist: list(blocker.blacklist) }, f) atexit.register(lambda: save_on_exit(smart_blocker))提示/etc/iptables.blacklist需手动添加到开机脚本如/etc/rc.local中执行加载。atexit确保异常退出时状态不丢失但生产环境建议用systemd管理进程并配合Restartalways。4. 避坑指南那些让毕设答辩当场翻车的5个真实陷阱4.1 现象程序跑10分钟就内存暴涨到2GBtop显示python进程RSS飙升原因conn_state字典无限增长未定期清理超时IP。尤其在高流量环境下defaultdict会为每个新IP创建新键而last_seen过期后未删除。解决在主循环中每30秒执行一次清理# 在主检测循环中 if time.time() - last_cleanup 30: for ip in list(conn_state.keys()): if time.time() - conn_state[ip][last_seen] 120: # 2分钟无活动 del conn_state[ip] last_cleanup time.time()4.2 现象封禁了攻击IP但自己的SSH连接也被断掉原因iptables规则插入顺序错误。iptables -I INPUT默认插到链首若前面有-p tcp --dport 22 -j ACCEPT规则新规则会插在它前面导致SSH也被DROP。解决明确指定插入位置或改用-A追加到末尾需配合-m state --state NEW# 正确在ACCEPT规则后插入 sudo iptables -I INPUT 5 -s 1.1.1.1 -j DROP # 或更稳妥用REJECT替代DROP避免TCP连接hang住 sudo iptables -I INPUT -s 1.1.1.1 -j REJECT --reject-with tcp-reset4.3 现象在CentOS 7上运行报错OSError: [Errno 19] No such device原因AF_PACKETsocket需要root权限且网卡名非eth0可能是ens33或enp0s3。socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.htons(0x0003))绑定时指定网卡名错误。解决动态获取主网卡名import subprocess def get_main_iface(): result subprocess.run([ip, route], capture_outputTrue, textTrue) for line in result.stdout.splitlines(): if default via in line: return line.split()[4] # 第5字段是网卡名 return eth04.4 现象检测到SYN Flood但get_slope()始终返回0.0原因_count_syn_recv_in_stream()统计逻辑错误——它把每个IP的syn_sent端口数累加但SYN Flood是单IP发海量SYN应统计len(conn_state[ip][syn_sent])的最大值而非总和。解决修正统计逻辑def _count_syn_recv_in_stream(self) - int: max_pending 0 now time.time() for ip, state in self.conn_state.items(): if (state[syn_sent] and now - state[last_seen] 2): max_pending max(max_pending, len(state[syn_sent])) return max_pending4.5 现象netfilterqueue安装后import报错ImportError: libnetfilter_queue.so.1: cannot open shared object file原因系统缺少libnetfilter_queue库pip install netfilterqueue只装Python包不装C依赖。解决# Ubuntu/Debian sudo apt-get install libnetfilter-queue-dev libnetfilter-queue1 # CentOS/RHEL sudo yum install libnetfilter_queue-devel libnetfilter_queue # 然后重新pip install pip uninstall netfilterqueue pip install netfilterqueue5. 生产就绪技巧用systemd托管、日志分级、以及那个救急的“后悔药”开关5.1 用systemd替代nohup实现进程守护与日志归档把python defender.py扔进nohup是学生做法。生产环境必须用systemd管理自动重启、资源限制、日志轮转。创建/etc/systemd/system/tcp-defender.service[Unit] DescriptionTCP Intrusion Detection System Afternetwork.target [Service] Typesimple Userroot WorkingDirectory/opt/tcp-defender ExecStart/usr/bin/python3 /opt/tcp-defender/defender.py Restartalways RestartSec10 LimitNOFILE65536 StandardOutputjournal StandardErrorjournal SyslogIdentifiertcp-defender [Install] WantedBymulti-user.target启用服务sudo systemctl daemon-reload sudo systemctl enable tcp-defender.service sudo systemctl start tcp-defender.service # 查看日志自动按天轮转 sudo journalctl -u tcp-defender -f好处Restartalways保证崩溃后10秒内恢复LimitNOFILE防文件描述符耗尽journalctl日志自带时间戳和优先级可-p err只看错误。5.2 日志分级DEBUG只存本地WARN以上推送到syslog检测系统日志要分层DEBUG每包解析详情只写/var/log/tcp-defender/debug.log每日轮转INFO封禁动作、状态变更写/var/log/tcp-defender/info.logWARNING/ERROR规则冲突、socket异常走syslog触发rsyslog转发到SIEM。import logging from logging.handlers import RotatingFileHandler def setup_logger(): # DEBUG日志 debug_handler RotatingFileHandler( /var/log/tcp-defender/debug.log, maxBytes10*1024*1024, # 10MB backupCount5 ) debug_handler.setLevel(logging.DEBUG) # INFO日志 info_handler RotatingFileHandler( /var/log/tcp-defender/info.log, maxBytes5*1024*1024, backupCount3 ) info_handler.setLevel(logging.INFO) # WARNING走syslog syslog_handler logging.handlers.SysLogHandler(address/dev/log) syslog_handler.setLevel(logging.WARNING) logger logging.getLogger(tcp-defender) logger.setLevel(logging.DEBUG) logger.addHandler(debug_handler) logger.addHandler(info_handler) logger.addHandler(syslog_handler) return logger logger setup_logger() logger.info(Defender started on %s, get_main_iface()) logger.warning(SYN Flood detected: slope%.2f, dos_detector.get_slope())5.3 “后悔药”开关一键暂停所有封禁保留检测但不解封答辩现场老师说“封错了怎么办”你不能说“重启”。必须有热停用开关创建/tmp/tcp-defender.pause空文件作为暂停标记主循环中检查该文件存在则跳过block_ip()调用但继续记录日志提供defender-pause.sh和defender-resume.sh脚本。# defender-pause.sh #!/bin/bash touch /tmp/tcp-defender.pause echo Paused. Current blocked IPs: iptables -L INPUT -v -n | grep DROP | grep -E [0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}# 在封禁逻辑前加检查 def block_ip(self, ip: str): if os.path.exists(/tmp/tcp-defender.pause): logger.info(Blocked paused. Would block %s, ip) return # 原封禁逻辑...真实教训我在某次课程设计答辩时老师故意用nmap -sS localhost触发封禁我当场执行./defender-pause.sh然后说“老师您看现在它还在检测、还在记日志只是不执行封禁——这就是我们设计的熔断机制。” 答辩直接过了。技术细节可以补但这种面向真实场景的思考才是毕设的灵魂。希望帮到你。本文还有配套的精品资源点击获取