一、引言为什么还要自己写端口扫描器做资产测绘、渗透测试或者内网巡检时端口扫描几乎是第一步。市面上已经有 nmap、masscan、naabu 这些成熟工具功能强大、速度极快那为什么还要用 Python 自己造轮子原因有三点。第一是可编程性nmap 的输出格式再灵活也很难直接嵌入到你自己的调度系统、CMDB 或者自动化报告流水线里而自己写的扫描器天然就是一个 Python 函数结果就是内存里的数据结构。第二是可控性你需要控制并发节奏、超时策略、扫描顺序甚至要根据目标的防护策略动态调整速率这些在通用工具里往往要靠复杂的参数组合。第三是理解深度只有亲手处理过ECONNREFUSED、ETIMEDOUT、TIME_WAIT这些东西你才会真正明白 nmap 输出里 “closed” 和 “filtered” 的区别意味着什么。本文从最基础的 TCP 全连接扫描讲起逐步实现一个支持并发、Banner 识别和结果分析的扫描器并给出实战中容易踩的坑和优化建议。补充本文适合谁如果你已经会写socket.connect()但不确定connect_ex的返回值到底代表什么如果你用过多线程但被TIME_WAIT搞崩过如果你想把扫描结果直接喂给下游的漏洞验证模块——那么这篇文章就是为你准备的。我们会从最朴素的 20 行代码开始每一步都解释“为什么这么写”而不是直接丢给你一个“最佳实践”。最终你会得到一个可以直接放进工具箱的portscanner.py它支持端口范围解析、线程池并发、Banner 抓取、JSON 输出并且能在 Linux/macOS/Windows 上稳定运行。二、核心原理connect 扫描到底发生了什么2.1 三次握手与端口状态判定TCP 全连接扫描nmap 中的-sT的原理极其朴素调用connect()系统调用观察返回值。端口开放目标回 SYNACK本机回 ACK三次握手完成connect()返回 0。端口关闭目标回 RSTconnect()立即返回错误码ECONNREFUSEDLinux 下为 111。被过滤目标或中间防火墙直接丢弃 SYN 包没有任何响应connect()一直阻塞直到超时。这三者的区分是扫描器全部价值的基础。注意closed和filtered的信息量完全不同前者说明主机活着但服务没起后者说明存在包过滤设备往往意味着更严格的安全策略。2.1.1 深入 TCP 状态机从 SYN 到 RST为了真正理解connect_ex的返回码我们需要回顾 TCP 状态转换。当客户端调用connect()时内核会发送一个 SYN 包并将套接字置于SYN_SENT状态。接下来有三种可能收到 SYNACK内核发送 ACK套接字进入ESTABLISHEDconnect()返回 0。此时目标端口处于监听状态应用层可以接受连接。收到 RST内核将错误码设置为ECONNREFUSEDconnect_ex返回该错误码。这意味着目标主机可达但对应端口没有进程监听或者监听队列已满。RST 包是 TCP 协议中用于拒绝连接的标准响应。没有任何响应内核会进行 SYN 重传Linux 默认重试 6 次约 127 秒但connect()的超时由用户空间的setsockopt(SO_SNDTIMEO)或settimeout()控制。一旦超时Python 抛出socket.timeoutPython 3.10 为TimeoutError。这通常意味着中间有防火墙丢弃了 SYN 包或者目标主机不存在。不同操作系统返回的错误码不同。Linux 下ECONNREFUSED是 111Windows 下是 10061。connect_ex返回的是errno值而不是异常。因此我们判断“关闭”时最好直接判断err ! 0且不是超时而不是硬编码 111。但如果你想区分“关闭”和“其他错误”如网络不可达可以引入errno模块importerrnoiferrerrno.ECONNREFUSED:returnport,closed,rtteliferrerrno.EHOSTUNREACH:returnport,unreachable,rtt2.1.2 超时设置的陷阱settimeout()设置的是阻塞套接字的超时时间。它同时影响connect()、recv()和send()。在扫描场景中我们通常只关心connect的超时。但要注意如果目标主机不存在DNS 解析可能先消耗几秒如果目标在公网且丢包严重1 秒的超时可能太短导致大量误报filtered。经验值是内网 0.5–1 秒公网 2–3 秒。对于关键目标可以先用ping或 ICMP 探测主机存活再对存活主机进行端口扫描避免对大量不存在的 IP 浪费时间。2.2 全连接扫描 vs 半开扫描半开扫描SYN Scan只发 SYN收到 SYNACK 后直接发 RST 断开不完成握手因此更快、更隐蔽也不会在目标的应用日志里留下完整连接记录。但它需要自己构造原始套接字依赖 root 权限和scapy/libpcap复杂度陡增。全连接扫描的优势是只用标准库、不需要特权、结果可靠。代价是会在目标侧产生完整的连接日志且本机会积累大量TIME_WAIT。作为基础实现我们选它。2.2.1 半开扫描的简单实现仅供理解如果你确实需要半开扫描可以用scapy快速实现fromscapy.allimportsr1,IP,TCP,RandShortdefsyn_scan(host,port,timeout1):src_portRandShort()pktIP(dsthost)/TCP(sportsrc_port,dportport,flagsS)respsr1(pkt,timeouttimeout,verbose0)ifrespisNone:returnfilteredifresp.haslayer(TCP):ifresp[TCP].flags0x12:# SYNACK# 发送 RST 断开sr1(IP(dsthost)/TCP(sportsrc_port,dportport,flagsR),timeout1,verbose0)returnopenelifresp[TCP].flags0x14:# RSTACKreturnclosedreturnunknown这段代码需要 root 权限且不同网络环境下行为差异很大。对于生产环境建议直接使用masscan或nmap -sS自己写半开扫描更多是为了学习 TCP/IP 协议栈。本文后续仍以全连接扫描为主因为它跨平台、无需特权且结果足够可靠。2.3 并发模型的选择端口范围 1–65535如果每个端口超时 1 秒串行扫描最坏情况要 18 小时。所以并发是必须的。可选方案方案优点缺点多线程 阻塞 socket代码简单兼容性最好线程开销大1000 以上收益递减selectors/select多路复用单线程资源占用低手写状态机代码复杂asyncio语法优雅轻松上万并发对同步库不友好调试稍难本文给出线程池和 asyncio 两个版本前者用于理解原理后者用于生产。2.3.1 并发模型性能对比为了让你有直观感受我在一台 4 核 8G 的 Linux 虚拟机上对127.0.0.1的 1–10000 端口进行了扫描超时 0.5 秒结果如下方案并发数耗时内存占用备注串行1约 5000 秒极低仅用于教学线程池200约 3.2 秒约 40 MB受 GIL 影响但 IO 密集场景足够asyncio1000约 1.8 秒约 25 MB单线程事件循环开销更小selectors1000约 1.5 秒约 20 MB代码最复杂但性能最好注意扫描本地回环地址时所有端口都会立即返回 RST 或 SYNACK几乎没有超时所以耗时极短。扫描公网目标时瓶颈在于网络延迟和超时并发数需要根据目标网络质量调整。线程池的max_workers不宜超过 500否则线程切换开销反而会降低吞吐量。asyncio 则可以轻松支持 1000–5000 个并发连接。三、实战从 20 行到可用的扫描器3.1 最小可用版本单端口探测importsocketimporttimedefscan_port(host:str,port:int,timeout:float1.0):返回 (port, state, rtt_ms)starttime.perf_counter()socksocket.socket(socket.AF_INET,socket.SOCK_STREAM)sock.settimeout(timeout)try:errsock.connect_ex((host,port))rtt(time.perf_counter()-start)*1000iferr0:returnport,open,rtt# ECONNREFUSED111(Linux) / 10061(Windows) 说明收到了 RSTreturnport,closed,rttexcept(socket.timeout,TimeoutError):# 3.10 起 socket.timeout 即 TimeoutError 的别名returnport,filtered,timeout*1000exceptOSErrorase:returnport,error,0.0finally:sock.close()if__name____main__:print(scan_port(127.0.0.1,22))这里的关键细节是connect_ex与connect的行为差异connect_ex在收到 RST 时返回错误码而不抛异常但在超时时仍然会抛出socket.timeout。所以两种分支都要处理否则超时目标会直接让程序崩溃。3.1.1 逐行解析与常见疑问socket.AF_INET指定 IPv4 地址族。如果要支持 IPv6需要改为AF_INET6并且地址格式也要变化。socket.SOCK_STREAM指定 TCP 协议。如果是 UDP 扫描要用SOCK_DGRAM但 UDP 的端口状态判定逻辑完全不同没有握手只能靠 ICMP 端口不可达或应用层响应。sock.settimeout(timeout)设置阻塞操作超时。注意这个超时是“每次操作”的超时不是整个扫描的总超时。connect_ex返回errno0 表示成功。非零值需要结合errno模块判断具体错误。time.perf_counter()提供最高可用分辨率的单调时钟适合测量短时间间隔。finally: sock.close()确保套接字被关闭避免文件描述符泄漏。在高并发扫描中忘记关闭套接字会导致OSError: [Errno 24] Too many open files。常见疑问为什么不用sock.connect()然后捕获异常因为connect()在收到 RST 时抛出ConnectionRefusedError在超时时抛出socket.timeout在其它错误时抛出OSError。用connect_ex可以统一处理返回值代码更简洁。但要注意connect_ex在超时时仍然会抛异常所以异常处理不可省略。3.2 并发版本线程池 参数解析importargparseimportsocketimportstructimporttimefromconcurrent.futuresimportThreadPoolExecutor,as_completed COMMON_PORTS[21,22,23,25,53,80,110,135,139,143,443,445,993,995,1433,1521,3306,3389,5432,6379,8080,8443]defparse_ports(spec:str):portsset()forpartinspec.split(,):partpart.strip()if-inpart:a,bpart.split(-,1)ports.update(range(int(a),int(b)1))elifpart:ports.add(int(part))returnsorted(pforpinportsif0p65536)defscan_port(host,port,timeout):starttime.perf_counter()socksocket.socket(socket.AF_INET,socket.SOCK_STREAM)sock.settimeout(timeout)# 主动关闭时发 RST 而非 FIN避免海量 TIME_WAIT 耗尽本地端口sock.setsockopt(socket.SOL_SOCKET,socket.SO_LINGER,struct.pack(ii,1,0))try:errsock.connect_ex((host,port))rtt(time.perf_counter()-start)*1000returnport,(openiferr0elseclosed),rttexcept(socket.timeout,TimeoutError):returnport,filtered,timeout*1000exceptOSError:returnport,error,0.0finally:sock.close()defmain():parserargparse.ArgumentParser(descriptionTCP端口扫描器)parser.add_argument(host,help目标主机如 192.168.1.1 或 example.com)parser.add_argument(-p,--ports,default,.join(map(str,COMMON_PORTS)),help端口范围如 22,80,443 或 1-1000)parser.add_argument(-t,--timeout,typefloat,default1.0,help超时秒数)parser.add_argument(-w,--workers,typeint,default200,help并发线程数)argsparser.parse_args()portsparse_ports(args.ports)print(f[*] 开始扫描{args.host}端口数{len(ports)}并发{args.workers})start_timetime.perf_counter()results[]withThreadPoolExecutor(max_workersargs.workers)asexecutor:futures{executor.submit(scan_port,args.host,port,args.timeout):portforportinports}forfutureinas_completed(futures):port,state,rttfuture.result()results.append((port,state,rtt))ifstateopen:print(f[] 端口{port}开放 ({rtt:.1f}ms))elapsedtime.perf_counter()-start_timeprint(f[*] 扫描完成耗时{elapsed:.2f}秒)open_ports[rforrinresultsifr[1]open]print(f[*] 开放端口数:{len(open_ports)})forport,_,rttinsorted(open_ports):print(f{port}/tcp open{rtt:.1f}ms)if__name____main__:main()3.2.1 代码关键点解析SO_LINGER设置struct.pack(ii, 1, 0)表示开启 linger超时时间为 0。当close()被调用时内核会立即发送 RST 并丢弃发送缓冲区而不是正常的四次挥手。这能有效避免扫描器在短时间内产生大量TIME_WAIT状态的连接。但要注意RST 关闭可能导致对端收到Connection reset by peer对于需要优雅关闭的场景如抓取 Banner 后应慎用。在纯端口扫描中我们只关心连接是否建立不传输应用层数据所以用 RST 关闭是安全的。parse_ports函数支持逗号分隔的单个端口和范围例如22,80,443,8000-8100。内部使用set去重最后排序。过滤掉 0 和 65536 以上的无效端口。这个函数虽然简单但非常实用你可以把它复制到任何需要解析端口的地方。线程池大小默认 200。对于内网扫描200 线程通常能在几秒内扫完 1000 个端口。如果目标网络延迟高可以适当降低到 100避免超时堆积。如果扫描公网大量 IP建议每个 IP 使用 50–100 线程并控制总并发。结果收集使用as_completed按完成顺序处理结果而不是按提交顺序。这样开放端口会尽快打印出来用户体验更好。但要注意future.result()可能抛出异常虽然我们的scan_port已经捕获了大部分异常但为了健壮性可以在外层再加一层try。3.2.2 运行示例假设保存为portscanner.py运行python portscanner.py192.168.1.1-p1-1000-w300-t0.5输出可能如下[*] 开始扫描 192.168.1.1端口数 1000并发 300 [] 端口 22 开放 (2.3 ms) [] 端口 80 开放 (1.8 ms) [] 端口 443 开放 (2.1 ms) [] 端口 3306 开放 (3.4 ms) [*] 扫描完成耗时 1.52 秒 [*] 开放端口数: 4 22/tcp open 2.3 ms 80/tcp open 1.8 ms 443/tcp open 2.1 ms 3306/tcp open 3.4 ms可以看到扫描 1000 个端口只用了 1.5 秒这就是并发的威力。3.3 Banner 识别让扫描结果更有价值仅仅知道端口开放是不够的。我们需要知道端口后面运行的是什么服务、什么版本。Banner 抓取Banner Grabbing就是向开放端口发送一个简单的请求或者直接读取然后分析返回的文本。很多服务在连接建立后会主动发送欢迎信息例如 SSH、SMTP、FTP、MySQL 等。下面是一个增强版的scan_port它在确认端口开放后尝试抓取 Bannerdefgrab_banner(host,port,timeout2.0):尝试抓取开放端口的 Banner返回字符串或 Nonetry:withsocket.create_connection((host,port),timeouttimeout)assock:sock.settimeout(timeout)# 有些服务需要客户端先发送数据才会响应这里发送一个换行# 但对于 SSH、SMTP 等它们会主动发送 Banner所以先尝试接收try:bannersock.recv(1024)ifbanner:returnbanner.decode(utf-8,errorsignore).strip()exceptsocket.timeout:pass# 如果没收到尝试发送 HTTP HEAD 请求针对 HTTP 服务try:sock.sendall(bHEAD / HTTP/1.0\r\n\r\n)bannersock.recv(1024)ifbanner:returnbanner.decode(utf-8,errorsignore).strip()except(socket.timeout,OSError):passexceptOSError:passreturnNone然后在scan_port中当err 0时调用grab_banneriferr0:bannergrab_banner(host,port,timeout)returnport,open,rtt,banner注意Banner 抓取会额外消耗时间并且可能被目标记录为“可疑连接”。在扫描大量端口时可以只对开放端口进行 Banner 抓取或者只对常见端口如 21、22、25、80、443、3306进行。另外有些服务如 HTTPS的 Banner 是加密的无法直接读取需要 TLS 握手这超出了基础扫描器的范围。3.4 asyncio 版本更高并发的实现线程池虽然简单但在扫描 65535 个端口时200 个线程仍然可能成为瓶颈。asyncio提供了单线程事件循环可以轻松管理数千个并发连接。下面是一个基于asyncio的扫描器核心importasyncioimporttimeasyncdefscan_port_async(host,port,timeout):starttime.perf_counter()try:reader,writerawaitasyncio.wait_for(asyncio.open_connection(host,port),timeouttimeout)rtt(time.perf_counter()-start)*1000writer.close()awaitwriter.wait_closed()returnport,open,rttexceptasyncio.TimeoutError:returnport,filtered,timeout*1000exceptConnectionRefusedError:returnport,closed,(time.perf_counter()-start)*1000exceptOSError:returnport,error,0.0asyncdefmain_async(host,ports,timeout1.0,concurrency1000):semasyncio.Semaphore(concurrency)asyncdeflimited_scan(port):asyncwithsem:returnawaitscan_port_async(host,port,timeout)tasks[limited_scan(port)forportinports]resultsawaitasyncio.gather(*tasks)returnresults使用asyncio.Semaphore控制并发数避免一次性创建过多连接导致文件描述符耗尽。asyncio.open_connection内部使用非阻塞套接字性能远高于线程池。实测在扫描 10000 个端口时asyncio 版本比线程池版本快 30%–50%内存占用也更低。四、结果分析与输出扫描完成后我们需要对结果进行结构化处理。一个实用的扫描器应该支持多种输出格式方便后续集成。4.1 保存为 JSONimportjsondefsave_json(results,filename):data{target:host,timestamp:time.**更多硬核网安与AI工具包请扫码获取完整源码**time(),open_ports:[{port:p,state:s,rtt_ms:round(r,2),banner