如何可靠检测TCP SYN端口扫描无更多返回报文以重试获取缺失响应

问题核心原因 #

你遇到的永久阻塞本质是两个问题叠加：

• SYN探测报文或对应的响应报文在网络中丢失，导致待检测的ports列表永远无法清空，recv_from的无限循环会一直等待新报文
• 现有逻辑没有设置接收超时兜底，也没有和发送任务的完成状态做联动

具体修复方案 #

1. 新增状态跟踪与超时逻辑 #

• 用trio的事件对象标记所有发送任务已完成
• 所有发送任务结束后，启动最长等待计时器，超时后直接终止接收器，剩余未从ports中移除的就是无响应/丢包的端口
• 支持对剩余端口做重试逻辑

2. 修改后代码 #

import trio
import socket

PORTS = 65535
TASKS = 15
# 自定义最大接收等待时间，单位秒，根据网络环境调整
MAX_RECV_WAIT = 10

async def recv_from(sock, target, cancel_event):
    while not cancel_event.is_set():
        try:
            # 给recvfrom加单次超时，避免永久阻塞
            with trio.move_on_after(1):
                data, addr = await sock.recvfrom(1024)
                if addr[0] == target:
                    yield data
        except:
            continue

async def receiver(remaining_ports: list) -> None:
    sock = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_TCP)
    cancel_event = trio.Event()
    async for packet in recv_from(sock, '192.168.1.1', cancel_event):
        if not remaining_ports:
            break
        src, dest, flags = unpack(packet)
        if dest == 6969 and src in remaining_ports:
            remaining_ports.remove(src)
            if flags == 20:
                print(f"port {src}: closed")
            elif flags == 18:
                print(f"port {src}: open")
    cancel_event.set()
    sock.close()

async def sender(idx: int, sender_done_event: trio.Event, ports_segment: list) -> None:
    sock = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_RAW)
    sock.setsockopt(socket.IPPROTO_IP, socket.IP_HDRINCL, 1)
    for port in ports_segment:
        tcp_packet = build_tcp_packet(port)
        ipv4_packet = build_ipv4_packet()
        await sock.sendto(ipv4_packet + tcp_packet, ('192.168.1.1', port))
    sock.close()
    # 标记当前发送任务完成
    sender_done_event.set()

async def scan_round(scan_ports: list) -> list:
    """单次扫描，返回未收到响应的端口列表"""
    remaining_ports = scan_ports.copy()
    async with trio.open_nursery() as nursery:
        sender_done_events = []
        # 拆分端口段给多个发送任务
        segment_size = len(scan_ports) // TASKS
        for i in range(TASKS):
            start = i * segment_size
            end = start + segment_size if i != TASKS-1 else len(scan_ports)
            seg = scan_ports[start:end]
            event = trio.Event()
            sender_done_events.append(event)
            nursery.start_soon(sender, i, event, seg)
            await trio.sleep(0.5)
        # 启动接收器
        nursery.start_soon(receiver, remaining_ports)
        # 等待所有发送任务完成
        for event in sender_done_events:
            await event.wait()
        # 所有报文发完后，再等待MAX_RECV_WAIT秒收响应
        await trio.sleep(MAX_RECV_WAIT)
        # 取消所有任务，结束本次扫描
        nursery.cancel_scope.cancel()
    return remaining_ports

async def main() -> None:
    # 初始全端口扫描
    remaining = list(range(PORTS))
    retry_times = 2 # 自定义重试次数
    for i in range(retry_times + 1):
        if not remaining:
            break
        print(f"第{i+1}次扫描，待扫描端口数：{len(remaining)}")
        remaining = await scan_round(remaining)
    if remaining:
        print(f"多次扫描后无响应端口：{remaining}")

if __name__ == '__main__':
    trio.run(main)

3. 额外优化点 #

• 把ports.remove改为先判断端口是否在列表中再删除，避免重复收到响应时报错
• 拆分单次扫描逻辑，可直接对无响应端口做多次重试
• 发送速率可根据网络环境调整，避免短时间发送报文过多被防火墙拦截导致丢包

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