如何使用多线程检测SYN数据包，实现Python端口扫描检测？

首先要明确：使用原始套接字时，不需要为每个端口创建线程或绑定端口。原始套接字（SOCK_RAW）搭配IPPROTO_TCP可以捕获所有经过本机的TCP数据包，无论目标端口是多少。你的单端口代码里绑定80端口是多余的操作，反而可能因为端口被占用导致报错。

核心思路 #

1. 单个原始套接字负责捕获所有TCP数据包，这是最高效的方式（多线程抓包会导致重复捕获相同数据包）。
2. 如果需要提升数据包处理效率，可采用「单线程抓包 + 多线程处理」的模式：抓包线程将数据包放入队列，多个工作线程从队列取出并解析检测。
3. 解析TCP头部的目标端口和SYN标志位，判断是否为端口扫描的SYN包。

改进后的代码示例 #

基础版（单线程，检测所有端口SYN包） #

import socket
import struct

def main():
    # 创建原始套接字，捕获所有TCP数据包
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_TCP)
        sock.setblocking(False)
    except PermissionError:
        print("需要以管理员/root权限运行")
        return
    except socket.error as e:
        print(f"套接字创建失败: {e}")
        return

    print("监听所有端口的SYN数据包...")
    while True:
        try:
            data, addr = sock.recvfrom(65535)  # 接收最大可能的数据包
        except BlockingIOError:
            continue
        except socket.error as e:
            print(f"接收数据错误: {e}")
            continue

        # 解析IP头部（兼容含选项的变长IP头）
        ip_header = data[:20]
        iph = struct.unpack('!BBHHHBBH4s4s', ip_header)
        tcp_offset = (iph[0] & 0xF) * 4  # 计算IP头部实际长度
        tcp_header = data[tcp_offset:tcp_offset+20]

        # 解析TCP头部
        tcph = struct.unpack('!HHLLBBHHH', tcp_header)
        dest_port = tcph[1]  # 提取目标端口
        flags = tcph[5]      # 提取TCP标志位

        # 检测SYN标志（0x02）
        if flags & 0x02:
            print(f"检测到SYN包 -> 来源IP: {addr[0]}, 目标端口: {dest_port}")

if __name__ == "__main__":
    main()

多线程处理版（抓包+线程池并行解析） #

如果数据包量很大，单线程解析可能跟不上，可以用线程池来并行处理：

import socket
import struct
import concurrent.futures
from queue import Queue
import threading

def process_packet(data_queue):
    while True:
        data, addr = data_queue.get()
        if data is None:  # 线程退出信号
            break

        try:
            # 解析IP头部
            ip_header = data[:20]
            iph = struct.unpack('!BBHHHBBH4s4s', ip_header)
            tcp_offset = (iph[0] & 0xF) * 4
            tcp_header = data[tcp_offset:tcp_offset+20]

            tcph = struct.unpack('!HHLLBBHHH', tcp_header)
            dest_port = tcph[1]
            flags = tcph[5]

            if flags & 0x02:
                print(f"SYN包检测 -> 来源: {addr[0]}, 目标端口: {dest_port}")
        except Exception as e:
            print(f"数据包解析错误: {e}")
        finally:
            data_queue.task_done()

def main():
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_TCP)
        sock.setblocking(False)
    except PermissionError:
        print("需要管理员/root权限运行")
        return

    # 创建队列传递数据包，避免阻塞抓包线程
    data_queue = Queue(maxsize=1000)
    # 启动3个并行处理线程
    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
        for _ in range(3):
            executor.submit(process_packet, data_queue)

        print("监听中...")
        try:
            while True:
                try:
                    data, addr = sock.recvfrom(65535)
                    data_queue.put((data, addr))
                except BlockingIOError:
                    continue
        except KeyboardInterrupt:
            print("\n停止监听")
            # 发送退出信号给所有处理线程
            for _ in range(3):
                data_queue.put((None, None))

if __name__ == "__main__":
    main()

关键注意事项 #

• 权限要求：运行原始套接字必须拥有管理员/root权限，否则会抛出权限错误。
• 数据包解析：IP头部可能包含选项字段，需通过IP头部的长度字段计算实际偏移量，不能直接固定为20字节。
• 避免重复抓包：不要为每个端口创建原始套接字，单个套接字即可捕获所有TCP包，多套接字会导致重复处理相同数据包。
• 性能优化：普通流量场景下单线程足够；高流量场景可考虑进程池（规避Python GIL对CPU密集型解析的影响）。

内容的提问来源于stack exchange，提问作者Yilerii