Python3综合扫描工具源码解析：从端口扫描到漏洞检测的自动化实践

简介：一份基于 Python3 的多功能网络安全扫描工具源码，面向企业安全自检、授权渗透测试与安全开发学习者。项目集敏感文件探测、WAF/CDN 识别、端口扫描、服务指纹识别、操作系统检测、弱口令检测、漏洞利用扫描、SQL 注入检测、CDN 绕过与旁站查询于一体，并预置常见中间件未授权访问和已知漏洞的 POC 脚本，便于直接验证目标资产安全性。压缩包共 43 个文件、约 6.99MB，32 个 Python 脚本构成核心检测逻辑，配合 Markdown 说明、JSON 配置、文本报告、GeoLite2 数据库与许可证文件，目录划分明确，适合二次开发与自定义检测模块。目前已有 398 人学习参考。通过源码可梳理扫描器整体设计思路，掌握各检测项的实现细节，快速搭建属于自己的轻量级安全检测工具；需注意该工具仅限合法授权场景使用，禁止用于未授权入侵。

1. 综合扫描工具不是工具合集：这套 Python3 源码到底解决什么问题

如果你接手过几十台机器的内网巡检，一定经历过这种场景：先用 Nmap 扫端口，再用脚本批量探测弱口令，最后手动核对服务版本去查 CVE。单看每一步都不难，可一旦 IP 上百、服务类型一多，碎片化的命令和输出格式会直接把人耗死。基于 Python3 的综合网络安全扫描工具设计源码，要解决的正是这个问题——不是再造一个 Nmap，而是把端口发现、服务识别、漏洞匹配、弱口令检测、报告输出串成一条自动化流水线。

这类工具的核心价值在于信息搜集的“链路梳理”：让每个环节的输出成为下一个环节的输入，并统一汇总成结构化结果。适合运维、安全测试和开发自检场景。入门者可以用它理解扫描器的工作机制，有经验的人则更关心模块怎么拆分、并发怎么控制、误报怎么收敛。

2. 源码怎么拆：模块边界、数据流与库选型

2.1 先定数据流，再写代码：从目标输入到报告输出的管线

很多半吊子的扫描工具拿到手没法改，根子是模块边界没划清楚。我的习惯是先画一条数据流，再按数据流切模块。

这套 Python3 扫描工具的典型数据流是：目标列表（IP/网段/域名）→ 端口发现 → 服务指纹识别 → 漏洞规则匹配 → 弱口令/专项检测 → 结果聚合 → 报告导出。每一步必须只依赖上一步的结果，不能越层调用。比如弱口令检测不该自己重新去做端口发现，而是接收“IP+端口+服务类型”的结构化输入。

模块化之后，新增检测能力就变成“加一个插件”，而不是改主流程。这种设计对源码工程有三个直接好处：排错定位容易、单模块可独立调试、新规则可以灰度上线。如果你打算在别人的源码基础上二次开发，先别急着看代码，把 main 函数里每个子功能之间的数据交接点标出来，通常比逐行读代码更能理解设计意图。

2.2 Python3 库选型：不同层的组件如何取舍

选型这块,我踩过不少坑。综合扫描工具用到的库按职责可以分为五层：网络连接层、协议解析层、并发调度层、漏洞匹配层、输出序列化层。

网络连接层我一般直接用标准库 socket，配合 timeout 参数。追求更快可以上 scapy 发 SYN 半开扫描，但 scapy 需要 root/管理员权限，在 Windows 上装 Npcap 还容易被安全软件拦截，作为综合工具的底层协议库合适，作为默认扫描模块不太合适。Nmap 的 python3-nmap 库本质是调外部 nmap 二进制，适合作为“外部工具调用适配层”，但不应该成为整套源码的基座，否则脱离了独立扫描能力。

协议解析层常用 paramiko（SSH）、python-redis-lib（Redis）、requests/urllib3（HTTP）。需要提一句 Python3 在 cryptography 依赖上的安装，很多 CentOS 7 环境装 paramiko 会挂在 cryptography 编译上，后面避坑章细说。

并发调度层不要自己在 socket 循环里塞 thread，用 concurrent.futures 的 ThreadPoolExecutor 足够，可控性比 asyncio 好，学习成本低。扫描场景是 IO 密集型，协程收益不明显，多进程除了耗内存没有额外收益。

漏洞匹配层不要依赖内存中的大字典，用 JSON/YAML 规则文件加载，每条规则包含指纹关键字和端口条件。输出层用 json 做程序间交换，用 Markdown 或 CSV 做人读报告。

2.3 日志、结果与临时状态：三类数据不要混在一起

源码工程里最常见的问题是把调试信息、检测结果、中间状态一股脑写进同一个变量或文件。这个工具的设计里我建议分成三路：logger 只负责运行日志，控制台和文件分开；result 对象只存最终命中的检测证据；session 缓存只存临时的“已扫过/待扫”状态，比如已探测到开放端口，就不要再重复扫描。

输出格式上，每条命中的记录至少要包含：目标 IP、端口、协议、服务名、命中规则 ID、匹配证据原文、耗时。有了证据原文，后续人工复核才不用重新连一次目标。这也是判断一套扫描工具源码是否专业的隐性指标——光给结论不给证据，只能算玩具脚本。

到这里，框架考虑清楚了，下一章可以开始写核心模块的落地实现。

3. 从 socket 到服务识别：端口扫描和 Banner 抓取的落地实现

3.1 线程池控制的 TCP 全连接扫描最小实现

端口扫描是整套工具的地基，地基不稳后面全是幻觉。先给一个可独立运行的 TCP 全连接扫描模块，重点在超时控制和并发上限。

import socket from concurrent.futures import ThreadPoolExecutor, as_completed def scan_one(ip: str, port: int, timeout: float = 1.0) -> dict: """对单个端口做 TCP 全连接探测""" sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(timeout) result = {"ip": ip, "port": port, "state": "closed"} try: if sock.connect_ex((ip, port)) == 0: result["state"] = "open" except socket.error as e: # 抛异常不等于端口关闭，可能是网络不可达或超时 result["error"] = str(e) finally: sock.close() return result def scan_ports(ip: str, ports: list, timeout: float = 1.0, max_workers: int = 200) -> list: """对多端口并行探测，返回开放端口列表""" open_ports = [] with ThreadPoolExecutor(max_workers=max_workers) as executor: futures = {executor.submit(scan_one, ip, p, timeout): p for p in ports} for future in as_completed(futures): res = future.result() if res["state"] == "open": open_ports.append(res["port"]) return sorted(open_ports) if __name__ == "__main__": # 典型用法：先扫常见端口列表，再对全端口段做二次精确探测 common_ports = [21, 22, 23, 25, 53, 80, 135, 139, 443, 445, 1433, 1521, 3306, 3389, 5432, 6379, 8080, 8443] print(scan_ports("192.168.1.10", common_ports, timeout=1.5, max_workers=300))

逻辑说明：scan_one 用 connect_ex 而不是 connect，因为 connect_ex 在失败时返回出错码而不是抛异常，少一层异常捕获，可读性更好。timeout 参数要显示传，不要依赖操作系统默认的 TCP 超时，那样单端口可能阻塞几分钟。

参数说明：timeout 设 1.0 秒适合局域网，跨网段或运营商链路建议 2.0~3.0 秒；max_workers 在默认 200~500 之间，Windows 高并发时 socket 句柄容易耗尽，Linux 上 500 没问题，macOS 建议 200。全端口扫描（1~65535）建议先用 SYN 广播式粗扫再用这个模块精扫，直接 ThreadPoolExecutor 跑 65535 个端口容易触达对端 IDS 的封禁策略。

3.2 Banner 抓取：协议头才是服务识别的命根子

端口开不开放只是第一步，更关键的是识别端口后面跑的是什么。Banner 抓取的坑在于，不同协议交互方式差异极大，HTTP 要发 GET，SSH 只需要等待握手包，SMTP 要等欢迎语。统一实现是：先尝试连接后直接 recv，拿不到数据再按协议表定向发送探测载荷。

import socket def grab_banner(ip: str, port: int, timeout: float = 3.0) -> tuple: """抓取服务 banner，返回 (raw_banner, protocol_hint)""" probes = { 80: b"GET / HTTP/1.0\r\nHost: %s\r\n\r\n" % ip.encode(), 443: b"GET / HTTP/1.0\r\nHost: %s\r\n\r\n" % ip.encode(), 21: b"", 22: b"", 25: b"", } try: sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(timeout) sock.connect((ip, port)) if port in probes and probes[port]: sock.send(probes[port]) # 先收一次，收不到再收一次，避免拆包导致只拿到半个协议头 banner = sock.recv(256) if not banner: banner = sock.recv(256) return banner.decode("utf-8", errors="ignore"), guess_service(ip, port, banner) except Exception as e: return "", "unknown" finally: sock.close() def guess_service(ip: str, port: int, banner: bytes) -> str: """先按端口猜，再用 banner 特征覆盖""" port_hint = {3306: "mysql", 6379: "redis", 27017: "mongodb"} if port in port_hint: return port_hint[port] banner_lower = banner.lower() if b'ssh' in banner_lower: return "ssh" if b'220' in banner_lower and b'smtp' in banner_lower: return "smtp" return "unknown"

逻辑说明：先 recv 一次后判断空包再 recv 一次，是解决 TCP 粘包和半包的最土但最有效的办法。数据库类服务（Redis/MongoDB）通常不会主动发 banner，需要发送特定探测指令，识别逻辑要单独走协议探测队列，不能只靠被动 recv。

参数说明：recv 缓冲区 256 字节足够覆盖绝大多数协议的握手特征，过大反而拖慢收包速度。decode 必须用 errors="ignore"，不然遇到非 UTF-8 编码的二进制握手包会直接抛 UnicodeDecodeError。这个函数的结果会喂给后面的漏洞规则匹配层，所以 protocol_hint 的准确性直接影响漏洞命中率——服务识别翻车，后面全白搭。

3.3 与 Nmap 是互补，不是替代

这套源码里不用 python3-nmap 做全流程，但也不是完全不碰 Nmap。合理的设计是：默认基于 socket 的模块负责轻量快速摸底，结果输出到一个等待列表。当 socket 层识别出疑似高危端口组合（比如 22 端口开且 banner 显示老版本 OpenSSH）时，再调度 Nmap 的 -sV 参数做深度服务指纹精确识别。

互补策略能避免两个极端：一个是纯脚本扫描的指纹粗糙，另一个是全靠 Nmap 外部二进制导致 Python3 环境部署失败。如果你改源码时发现某个服务指纹总是识别错，优先检查 guess_service 的规则顺序，不是先去调扫描超时——这是我的血泪经验。

4. 漏洞探测与弱口令检测：规则引擎和爆发的边界控制

4.1 CVE 匹配不能硬编码：JSON 规则文件驱动

漏洞模块最忌讳把 CVE 编号和指纹硬编码在 Python 源码里。规则一多，改一次就要动一次代码，发版成本高。我常用的设计是两层：外层 JSON 规则文件描述匹配条件，内层一个 evaluator 函数负责解释执行。

import json import re class VulnRuleEngine: def __init__(self, rule_path: str): with open(rule_path, encoding="utf-8") as f: self.rules = json.load(f) def match(self, service: dict) -> list: """service 结构: {"ip":..., "port":..., "protocol":..., "banner":...}""" hits = [] banner = service.get("banner", "") for rule in self.rules: # 端口不符合直接跳过 if rule["port"] != service["port"] and rule["port"] != "any": continue expr = rule.get("banner_regex", "") if expr and re.search(expr, banner, re.IGNORECASE): hits.append(rule) return hits # 规则文件示例: vuln_rules.json # [{"id": "CVE-2019-0708-1", "port": 3389, "banner_regex": None, # "descript": "RDP service exposed, further check required"}]

逻辑说明：规则引擎里最关键的不是规则数量，而是匹配条件的可组合性——端口、banner 正则、协议类型三个字段按与的关系组合。实际使用时，对于没有 banner 特征的端口（如 RDP 3389），banner_regex 置空，规则只做端口命中，然后交给下一层专项检测。

参数说明：re.search 用 IGNORECASE 是为了兼容大小写不一致的协议头；规则文件里 port 字段支持 "any" 通配，避免为每个端口写重复规则。CVE 匹配是概率性命中，它做的是“疑似”标记，绝不能在报告里直接断言目标存在漏洞，证据链不足时要明确打上 need_verify 标签。

4.2 SSH 弱口令检测：paramiko 的调用参数与失败控制

弱口令检测在这类工具里争议较大，做的时候必须控制“爆发度”——即同一目标的尝试频次。这里给一个基于 paramiko 的最小实现，重点看怎么在一次连接里完成认证判断。

import paramiko def check_ssh_weak(ip: str, port: int, username: str, password: str, timeout: float = 5.0) -> bool: """尝试一组账号口令，返回是否成功""" client = paramiko.SSHClient() client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) try: client.connect( hostname=ip, port=port, username=username, password=password, timeout=timeout, allow_agent=False, look_for_keys=False, ) return True except paramiko.AuthenticationException: return False except Exception as e: # 超时/连接重置/服务端拒绝协议，都算检测不可用 return False finally: client.close()

参数说明：allow_agent=False 和 look_for_keys=False 很关键，它们禁止 paramiko 自动加载本机 SSH 私钥。否则在堡垒机上跑检测，可能会用你的私钥去试目标机器，那既不是弱口令检测，还可能给目标造成不可预期的后果。timeout 设 5 秒是折中值：太短在慢链路上误报“认证失败”，太长单组账号耗时拖垮整体进度。

弱口令检测模块一定要有“尝试次数上限”和“失败暂停”机制。我的经验是每 IP 最多尝试 10 组口令，连续失败 5 次后等待 30 秒。这既是防止账号被锁，也是对目标系统的基本尊重。

4.3 HTTP 服务的轻量检测与报告拼装

HTTP 服务检测要覆盖的是：HTTP 方法、响应头信息泄露、目录列举。这里不做复杂 DAST 扫描，只做信息收集级别的检测。

import requests def check_http_header(ip: str, port: int, timeout: float = 3.0) -> dict: """检查 HTTP 响应头中的安全相关字段""" url = f"http://{ip}:{port}/" result = {"url": url, "headers": {}, "findings": []} try: resp = requests.get(url, timeout=timeout, verify=False) result["status_code"] = resp.status_code result["headers"] = dict(resp.headers) server_header = resp.headers.get("Server", "") if not server_header: result["findings"].append("missing_server_header") if "X-Frame-Options" not in resp.headers: result["findings"].append("missing_x_frame_options") except requests.exceptions.RequestException as e: result["error"] = str(e) return result

逻辑说明：requests 库的 verify=False 在扫描场景下是必需的——目标站点证书大概率自签名或过期，如果让 SSL 校验阻断请求，检测就等于没做。这里只做三项检查，保持模块小而专，因为在综合工具里，HTTP 深度检测通常有专门的子模块承载。

参数说明：timeout 要同时覆盖连接和读取，requests 的 timeout=3.0 参数内部是两个超时的元组，单值意味着 connect 和 read 共用。在跨运营商链路时建议显式传 (3, 5)，不然 read 超时太短会导致漏报。

报告拼装在最后统一做，建议输出两种格式：JSON 给程序消费，Markdown 给人读。每一条 finding 都必须带上原始证据（响应头原值、banner 原文、URL），否则报告就失去审计价值。

5. 综合扫描工具避坑清单：误报、漏报与阻塞排查

5.1 线程池调得太大，扫描结果反而失真

现象：同一网段扫描时，把 max_workers 从 200 调到 1000，结果发现之前“开放”的端口大量变为“关闭”。

原因：线程数超过系统文件描述符上限时，socket() 或 connect() 直接抛 OSError，异常被吞掉后标记为 closed。另外目标主机也会因为连接风暴丢弃 SYN 或触发防扫描机制。

解决：先跑ulimit -n看当前进程的文件描述符上限，线程池大小严格小于该值的一半。跨网段扫描时用-sS方案代替全连接，或者干脆把并发降到 100 以内，宁可慢一点也不要得到一张假开放列表。

5.2 DNS 解析耗时阻塞了整个扫描主流程

现象：扫一个包含域名的主机列表，进度条卡在某处 30 秒不动，最后输出报告时发现时间全部耗在 socket.getaddrinfo 上。

原因：socket.connect 传入域名时会同步做 DNS 解析，在 DNS 服务器不可达时默认等待时间很长，Python3 标准库不好直接对解析阶段设超时。

解决：扫描前统一做一步“IP 化”，用一个 ThreadPoolExecutor 批量解析域名，并把解析失败的主机单独记入日志。之后全流程只处理 IP，避免在循环内部反复触发解析。这个前置步骤虽然简单，却能把大网段扫描时间缩短一半。

5.3 弱口令检测把目标账号锁了

现象：检测结束后，目标服务器管理员反馈部分账号被锁定，疑似暴力破解。

原因：并发检测模块没有做“单目标串行+失败退避”，多个插件（SSH、FTP、MySQL）同时打同一台机器的不同端口，每个模块各试 10 次，累计尝试次数远超锁定阈值。

解决：所有弱口令检测必须经过统一调度器，按“IP 维度串行”，跨协议尝试之间至少间隔 1 秒。规则库中尽量不用包含空密码的弱口令字典，那会直接导致无认证服务被锁。源码里建议加一个--lock-aware开关，开启时单 IP 总尝试次数硬编码为 8 次。

5.4 CentOS7 上 paramiko / cryptography 装不上

现象：pip install paramiko 时编译 cryptography 报错，显示缺少 rust 工具链或 OpenSSL 头文件。

原因：cryptography 新版本在 PyPI 上没有预编译 wheel，需要本地编译；而 CentOS7 自带的 OpenSSL 版本较旧，再叠加 python3.8 以下编译环境缺头文件，安装就变成了一场灾难。

解决：换用 openssl11-devel 和 python3-devel 后重新编译；更省事的做法是使用 ELRepo 或 SCL 源安装自带依赖的 Python 发行版，并固定 paramiko 版本到 2.x 而不是最新 3.x。还有一个土办法是连依赖一起离线下载 wheel 包放到内网 pip 源，这是没外网环境时的唯一解。

5.5 SSL 握手卡住 HTTP 端口检测

现象：检测 HTTPS 端口时，程序在 recv 阶段长时间阻塞，即使设置了 socket timeout 也偶尔失效。

原因：TLS 握手过程中，服务端在等待 ClientHello，客户端在等待服务端握手响应，双方都在等导致超时被协议栈误判；部分实现会在握手前先发一个 HTTP 明文请求，被服务端丢弃后无限等待。

解决：先用 recv 判断首个字节是否是 TLS 记录层类型（0x16），识别为 TLS 后先发一个标准 ClientHello 探测包。如果拿不到响应再放弃，不要同时发送 HTTP 明文探测。另外把 socket 超时从timeout改成(connect_timeout, read_timeout)两个独立参数，对 read 阶段单独设短超时，可以避免卡死现象反复出现。

6. 验证这套扫描源码：在本地复现、把误报率压低

拿到一套综合扫描工具源码，先别急着上生产网段，我一般会在本地起一个最小靶场验证。做法是用 Docker 跑几个特定版本的服务镜像，或者直接在本机开几个 Python HTTP 服务做自测。比如起一个python3 -m http.server 8080 --bind 127.0.0.1，再手动配置一个返回固定 Server 头的简易服务，用来验证端口扫描、Banner 抓取和规则引擎的联动。

先跑单目标扫描，对比实际开放的端口与报告输出，确认没有漏报；再故意配置一个“无响应”的端口，验证超时路径是否正确。第二步是多目标交叉验证：同一服务开在 3 台机器上，扫描结果必须一致，如果一台这周报 3379 端口开放、下周又不开放，多半是并发模块的瞬时抛错没有被捕获，要回头查 scan_one 里的超时异常分支。

字节校验上，我习惯在报告导出之前加一个去重和冲突排序的步骤：同一 IP 的相同 finding 只保留最高严重级别的证据；对协议识别为 unknown 的结果单独归档，不要混入正式命中。三个月迭代下来，我一个人维护这套工具，最大的体会是：误报率压到 20% 以下，工具才有人敢信；压不下去，再全的规则库都是黑匣子。最后补一个个人习惯——每次扫描前把输出目录按日期建好，结果文件和运行日志分开存。毕竟这行做久了，你总会需要回头翻三个月前的原始输出，到时候就知道后悔药有多难买。希望帮到你。

本文还有配套的精品资源，点击获取