端口扫描工具的实现：从零构建一个并发 TCP 端口扫描器

端口扫描是网络探测与安全审计的基石。本文不依赖现成的 Nmap，而是用 Python 从零实现一个并发 TCP 端口扫描工具，覆盖原理、架构、并发调度、服务指纹识别与可运行的完整代码，并配有原理与架构图。

目录

1. 端口扫描是什么
2. 原理基础：端口与 TCP 三次握手
3. 常见的扫描方式对比
4. 工具架构设计
5. 核心实现（Python 完整代码）
6. 扫描主流程
7. 性能与并发优化
8. 进阶方向
9. 总结

1. 端口扫描是什么

端口（Port）是传输层用来区分不同网络服务的逻辑编号，范围 0–65535。一台主机上可能同时运行 Web、SSH、数据库等多个服务，它们各自绑定一个端口。端口扫描的目的，就是探测目标主机上哪些端口处于开放状态，从而推断其上运行了哪些服务，常用于：

• 安全审计：检查自身网络对外暴露了哪些不必要的服务，缩小攻击面。
• 渗透测试：作为信息收集的第一步，为后续漏洞探测定位目标。
• 运维巡检：确认服务是否正常监听、端口是否被防火墙误拦。
• 网络资产盘点：摸清网段内主机与开放端口分布。

一个基本的端口扫描器，本质上是对目标每个端口发起一次连接尝试，并根据响应判定其状态。

2. 原理基础：端口与 TCP 三次握手

TCP 是面向连接的协议，建立连接需要经过三次握手（Three-way Handshake）。端口扫描正是利用握手过程来判断端口状态：

1. 客户端发送SYN报文，请求建立连接；
2. 服务端若在该端口监听，会回复SYN + ACK，表示"同意连接"；
3. 客户端再回复ACK，连接正式建立。

借助这一过程，我们可以在应用层通过一次connect()调用的结果来判定端口状态——这就是TCP Connect 扫描的核心原理：

判定的核心：connect()成功返回 → 端口开放；连接被拒绝或超时 → 端口关闭（Closed）或被防火墙过滤（Filtered）。

3. 常见的扫描方式对比

不同扫描方式在"是否完成完整握手、是否留下连接痕迹、是否需要系统权限"上各有取舍：

本文采用 TCP Connect 扫描——它不需要特殊系统权限，实现清晰，足以说明端口扫描工具的完整工程结构；半开扫描等进阶思路放在第 8 节。

4. 工具架构设计

把端口扫描器拆成清晰的模块，便于维护与扩展。整体分为四层：

• 配置 / CLI 层：解析命令行参数，确定目标主机、端口范围、超时与并发数。
• 扫描引擎：核心调度逻辑，负责并发分配端口探测任务，控制并发度与超时。
• 网络交互层：真正执行TCP Connect探测、读取服务 Banner 的底层操作。
• 结果处理：对探测结果做服务指纹识别与格式化输出（控制台 / JSON 文件）。

分层的好处是：换一种探测方式（如 SYN 半开）只需替换"网络交互层"，调度与输出逻辑完全复用。

5. 核心实现（Python 完整代码）

5.1 技术选型

• 语言：Python 3，标准库即可，零第三方依赖。
• 并发：concurrent.futures.ThreadPoolExecutor线程池，简单可靠。
• 目标解析：ipaddress库处理 IP / 网段，socket解析主机名。
• 超时：socket.settimeout()防止单端口探测卡死。

5.2 完整代码

#!/usr/bin/env python3""" port_scanner.py —— 一个并发 TCP 端口扫描器 用法: python port_scanner.py 192.168.1.1 python port_scanner.py 192.168.1.0/24 --ports 1-1000 python port_scanner.py example.com --ports 22,80,443,3306 --timeout 2 --json result.json """importargparseimportconcurrent.futuresimportipaddressimportjsonimportsocket# 常见服务端口 -> 服务名映射，用于快速标注已知端口COMMON_PORTS={21:"FTP",22:"SSH",23:"Telnet",25:"SMTP",53:"DNS",80:"HTTP",110:"POP3",135:"RPC",139:"NetBIOS",143:"IMAP",443:"HTTPS",445:"SMB",3306:"MySQL",3389:"RDP",5432:"PostgreSQL",6379:"Redis",8080:"HTTP-Alt",9200:"Elasticsearch",}defparse_targets(targets):"""解析目标：支持 IP、IP/掩码网段、主机名。返回 IP 字符串列表。"""result=[]fortintargets:if"/"int:# 网段，如 192.168.1.0/24result.extend([str(h)forhinipaddress.ip_network(t,strict=False).hosts()])else:try:result.append(str(ipaddress.ip_address(t)))# 单个 IPexceptValueError:result.append(socket.gethostbyname(t))# 主机名 -> IPreturnresultdefparse_ports(ports_spec):"""解析端口范围：支持 '1-1000'、'22,80,443' 或两者混合。"""ports=set()forpartinports_spec.split(","):part=part.strip()if"-"inpart:lo,hi=part.split("-")ports.update(range(int(lo),int(hi)+1))else:ports.add(int(part))returnsorted(ports)defgrab_banner(sock,timeout):"""读取服务 Banner，用于指纹识别。"""sock.settimeout(timeout)try:sock.sendall(b"\r\n")returnsock.recv(256).decode("utf-8",errors="ignore").strip()exceptException:returnNonedefprobe(host,port,timeout):"""对单个端口执行 TCP Connect 探测，返回结构化结果。"""sock=socket.socket(socket.AF_INET,socket.SOCK_STREAM)sock.settimeout(timeout)result={"port":port,"state":"closed","service":None,"banner":None}try:ifsock.connect_ex((host,port))==0:result["state"]="open"result["service"]=COMMON_PORTS.get(port,"unknown")result["banner"]=grab_banner(sock,timeout)sock.close()exceptsocket.timeout:result["state"]="filtered"# 超时，可能被防火墙丢弃sock.close()exceptException:sock.close()returnresultdefscan_host(host,ports,timeout,workers):"""并发扫描单个主机的一组端口，返回该主机的开放端口列表。"""open_ports=[]withconcurrent.futures.ThreadPoolExecutor(max_workers=workers)asex:futures={ex.submit(probe,host,p,timeout):pforpinports}forfutinconcurrent.futures.as_completed(futures):r=fut.result()ifr["state"]=="open":open_ports.append(r)returnhost,sorted(open_ports,key=lambdax:x["port"])defmain():parser=argparse.ArgumentParser(description="并发 TCP 端口扫描器")parser.add_argument("targets",nargs="+",help="目标：IP / 网段 / 主机名")parser.add_argument("--ports",default="1-1024",help="端口范围，如 1-1000、22,80,443")parser.add_argument("--timeout",type=float,default=1.0,help="单端口连接超时（秒）")parser.add_argument("--workers",type=int,default=100,help="并发线程数")parser.add_argument("--json",help="将结果输出到 JSON 文件")args=parser.parse_args()ips=parse_targets(args.targets)ports=parse_ports(args.ports)print(f"目标主机数:{len(ips)}，端口数:{len(ports)}，并发线程数:{args.workers}")all_results={}foripinips:host,opens=scan_host(ip,ports,args.timeout,args.workers)all_results[host]=opensforrinopens:line=f" [OPEN]{host}:{r['port']:<5}{r['service']}"ifr["banner"]:line+=f" banner={r['banner'][:60]}"print(line)ifargs.json:withopen(args.json,"w",encoding="utf-8")asf:json.dump(all_results,f,ensure_ascii=False,indent=2)print(f"结果已写入{args.json}")if__name__=="__main__":main()

5.3 关键实现要点

• connect_ex()vsconnect()：connect_ex()用返回码而非抛异常表示结果，0即连接成功，免去异常捕获的繁琐，适合高频探测。
• 并发模型：ThreadPoolExecutor把每个端口探测任务丢进线程池，as_completed一边收结果一边处理，主线程不阻塞。
• 超时兜底：settimeout()保证单端口最多等待timeout秒，避免因丢包导致整轮扫描卡死；超时统一归为filtered。
• Banner 识别：连接成功后发送一个空行或协议触发指令，读取服务返回的第一段文本，作为指纹线索（如 SSH 会回显SSH-2.0-OpenSSH_...）。
• 结果结构化：每条记录统一为{port, state, service, banner}，便于后续序列化成 JSON 或接入其他系统。

6. 扫描主流程

工具的完整执行流程自顶向下推进：

1. 输入目标：用户给定 IP、网段或主机名及端口范围；
2. 目标解析：把目标拆成 IP 列表 + 端口列表；
3. 并发调度：由线程池按并发度分发探测任务；
4. 端口探测：对每个端口执行 TCP Connect；
5. 结果判定：根据响应归为开放 / 关闭 / 过滤；
6. 服务指纹识别：对开放端口抓 Banner；
7. 结果输出：控制台打印或写入 JSON 文件。

每一步职责单一、边界清晰，新增功能（如存活主机探测、速率控制）可在对应步骤扩展。

7. 性能与并发优化

端口扫描是典型的 I/O 密集任务——大多数时间都花在等待网络响应上，CPU 几乎不参与。因此并发是提升扫描速度的关键。以扫描 1000 个端口为例，不同并发方式的耗时差异巨大：

• 单线程串行：逐个等待，1000 个端口即使全部秒回也要耗时数十秒，慢在网络往返的串行等待上；
• 多线程（ThreadPoolExecutor）：用线程池同时发起数百个连接，速度提升一个数量级；
• asyncio 协程：单线程内用事件循环管理海量 socket 连接，开销更低，在大规模扫描时进一步缩短耗时。

并发数如何取舍

• 并发过低：浪费带宽，扫描慢；
• 并发过高：触发目标防火墙限速、占用本地文件描述符（socket资源）耗尽，甚至造成网络拥塞。生产工具通常支持可配置并发数 + 速率限制（rate limiting），并加入随机延迟降低被检测概率。

8. 进阶方向

⚠️合规提醒：对未经授权的目标进行端口扫描属于探测行为，在多数地区可能违反法律法规。请务必只在自己拥有或有明确书面授权的网络范围内使用，或用于自身资产的合规巡检。

9. 总结

本文从端口扫描的原理出发，梳理了常见扫描方式，并用 Python 从零实现了一个结构清晰、可运行的并发 TCP 端口扫描器。核心要点：

• 原理：利用 TCP 三次握手的connect()结果判定端口开放/关闭/过滤；
• 架构：配置层、扫描引擎、网络交互层、结果处理四层分离，便于替换探测方式；
• 并发：I/O 密集任务用线程池/协程显著提速，需平衡并发度与稳定性；
• 指纹识别：抓取 Banner 推断服务，为安全审计提供更丰富的信息。

在此基础上，可以继续向 SYN 半开扫描、服务指纹库、异步 I/O 等方向演进，把它打磨成功能完整的网络探测工具。