基于Python的综合网络安全扫描工具：架构、源码与避坑实践

简介：基于Python3编写的多功能网络安全扫描工具源码包，适用于甲方自测或乙方授权安全评估场景，也适合安全初学者研究常见检测思路。压缩包共41个文件，约6.98MB，核心为31个Python脚本，覆盖敏感文件探测、WAF/CDN识别、端口扫描、服务识别、操作系统识别、弱口令检测、漏洞扫描、绕过CDN与旁站查询等检测模块；JSON配置文件用于维护扫描参数，txt文档提供使用说明与依赖清单，另有LICENSE许可协议和mmdb地址库辅助运行。这份源码将多种常用安全检查能力整合为可扩展的工具框架，目录按功能拆分，既有目标信息收集与主动扫描脚本，也包含Redis、Docker、Weblogic等未授权/漏洞利用检测示例，学习者可对照脚本梳理调用逻辑，也能直接改造用于授权评估。目前已有333人学习，适合需要快速搭建内部安全评估工具或希望从代码层面理解综合扫描器设计的人。

1. 为什么还要自己写一个Python网络安全扫描工具

看到“基于Python的综合网络安全扫描工具设计源码”这个标题，多数人第一反应是：Nmap、Masscan、OpenVAS不都很成熟了吗，为什么还要自己写一套？我的答案是：现成工具解决的是“能不能扫出来”，而自己写的源码解决的是“扫描之后怎么办、怎么跟自己的业务流程对接”。我在做内网资产梳理时，需要把扫描结果直接落到数据库、自动生成巡检报告、把弱口令和CVE列表按业务线分派给对应负责人，这些事靠手搓命令行再复制粘贴，不仅慢，还容易漏。

Python之所以适合干这个活，是因为它能把底层探测（socket、Scapy）、协议解析（构造HTTP、SNMP、SMB报文）和上层业务（结果入库、任务调度、报告生成）粘在一起，一份源码从探测到落地全链路打通。比如我们常见做法是先用异步并发做端口存活探测，再对存活端口做服务识别和CVE指纹匹配，最后用弱口令模块做合规检查。整个过程几百行代码就能搭出一个能跑的最小版本，这比想象中容易，也比想象中容易踩坑。

这个方向适合三类人：一是做等保或内网自查的安全工程师，需要批量发现资产和弱口令，并留下可追责的扫描记录；二是想往安全开发转的Python工程师，借这个项目把TCP状态机、HTTP指纹、并发调度都练一遍；三是需要在研发环境里做轻量合规检查的运维，不希望重工具占用太多资源。下面我会把整个工具的架构拆开，每一层都给出能直接复现的代码和参数解释，最后专门讲我调这套东西时翻车最狠的几个地方。

2. 先把工具拆成五个模块：架构设计决定你后面改代码的心情

综合扫描工具最容易犯的错是一上来就写一个巨型脚本，所有逻辑放在一个while循环里，最后连自己都维护不了。我一般会按数据流方向拆成五层：任务输入层、探测调度层、指纹与检测层、结果持久化层、报告输出层。这五层之间用统一的数据结构（比如字典或dataclass）传递结果，这样想给哪一层加功能都不会牵动全局。

2.1 模块划分与数据流：为什么我坚持用dataclass而不是dict

任务输入层负责接收IP段、域名列表、端口范围，比如192.168.1.0/24或10.0.0.1-10.0.0.50；探测调度层负责把一个大任务切成小任务并发执行；指纹与检测层拿到某个IP:Port之后，判断这是不是开放端口、跑的是什么服务、有没有已知CVE；结果持久化层把探测结果写进SQLite或MySQL；报告输出层生成HTML或CSV交给运营或合规部门。

这五层之间最关键的“语言”是结果对象。我最早图省事用dict传，结果字段名在各层之间越写越乱，有的地方叫port，有的地方叫service_port，改一处漏三处。后来统一改成dataclass，字段在定义时约束死，IDE还能自动补全，代码重构成本直线下降。

from dataclasses import dataclass, field from typing import Optional, List @dataclass class Target: ip: str port: int service: Optional[str] = None version: Optional[str] = None status: str = "closed" # open / closed / filtered cve_list: List[str] = field(default_factory=list) weak_passwords: List[str] = field(default_factory=list) raw_banner: Optional[str] = None

这个dataclass里status是探测层写进来的，service和version是服务识别层改写的，cve_list和weak_passwords是后面的检测模块追加的。各层之间只看这个对象，不看别的中间变量，出问题的时候直接打印一个Target就能定位是哪一步写坏了。

2.2 选型对比：自研探测 vs 调Nmap的边界在哪

有人会问：既然有python-nmap这个库，何必自己写socket探测？我用过的项目里，两种方案都有适用场景，做综合工具时它们经常同时存在。我习惯把“自研socket探测”作为第一层，因为它的并发控制、超时设置和输出格式最可控，且在授权扫描场景下不会扇出太多子进程占用服务器资源；而把Nmap作为可选的深度识别模块，当首轮探测发现端口开放但服务识别置信度不高时，才调Nmap做一次补充。

下面这个表是我在实际项目里总结的选型参考：

两种方式各占一半，不是非此即彼。综合工具的价值在于调度：先用自研探测把几千个IP:Port的首轮存活跑完（快，秒级），只对开放的端口和TCP服务标记进下一轮，再由Nmap或banner识别做深度检查。这样既保住了效率，也让后面的指纹识别有了靠谱的“素材”。

2.3 任务调度的根本问题：并发太高会被防火墙盯上，并发太低扫描慢成PPT

调度层我一般用ThreadPoolExecutor加信号量做双重控制，单目标并发和全局并发分开限制。比较稳的参数是：全局并发控制在200个任务以下，每个目标的端口并发控制在50以内；没有授权的边界上，我会再加一个asyncio.Semaphore，把总扫描速率限制到每秒500个包左右。

import asyncio from concurrent.futures import ThreadPoolExecutor class ScanScheduler: def __init__(self, max_workers=20, global_concurrency=200): self.executor = ThreadPoolExecutor(max_workers=max_workers) self.semaphore = asyncio.Semaphore(global_concurrency) self.rate_limiter = asyncio.Semaphore(50) # 每批最多50个探测任务 async def submit_scan(self, target: Target): async with self.semaphore: loop = asyncio.get_running_loop() result = await loop.run_in_executor( self.executor, self._probe_target, target ) return result def _probe_target(self, target: Target): # 实际探测逻辑在下一章实现，这里只做占位 return target

这里的max_workers控制线程池的大小，别设太大，否则GIL切换会吃掉大量CPU时间；global_concurrency控制的是跨目标的并发上限，主要帮你避免一轮扫描就打死网关。需要说明的是，信号量只是本地限速，对跨机器分布式部署的场景，还需要在外层做任务队列，这个后面第四章会提。

服务识别和漏洞检测都会拿到上一层的产物——一个大列表，里面每个元素是带有ip、port、status、banner的Target对象。只要这个数据流是干净的，后面每一个检测模块都只是“吃一个Target，吐一个Target”，组合起来就是一张完整的资产风险清单。

3. 端口与服务识别：把“有一个端口开着”升级为“这是一个SSH 7.4”

端口扫描的原理不神秘，就是向目标IP的某个端口发TCP SYN包或直接三次握手，看对方给不给出响应。但对于综合工具来说，知道“端口开了”只是第一步，更关键的是判断后面跑的服务是什么，因为漏洞检测、弱口令检测都是基于服务类型的。这一章我把端口扫描、服务识别、并发控制三件事连同源码一起讲透。

3.1 三种探测方式的取舍：TCP connect、TCP SYN、UDP探测

TCP connect就是直接用socket建立完整三次握手，成功即开放，实现最简单，但会在目标机器的日志里留下大量连接记录，治理严格的区域会告警。TCP SYN半开扫描需要构造SYN包并监听回应，效率高但需要root权限，而且许多云主机禁止裸socket操作。UDP探测更加麻烦，因为UDP无连接，只能靠ICMP端口不可达消息来判断，丢包率高，误报也多。

我在这套工具里的默认选择是TCP connect扫描，原因很实际：它在普通用户权限下就能跑，丢包率最低，结果稳定；在渗透测试或内网红队演练场景里，再考虑半开扫描。对每个端口设置1.5秒到3秒的超时，重试一次就够了，重试多了会拖慢整个任务调度。

import socket import concurrent.futures def tcp_connect_probe(ip: str, port: int, timeout: float = 2.0) -> bool: """TCP connect探测：成功返回True，失败返回False。""" sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(timeout) try: result = sock.connect_ex((ip, port)) return result == 0 finally: sock.close() def port_scan_batch(ip: str, ports: list, timeout: float = 2.0, workers: int = 100) -> list: """并发扫描一组端口，返回开放端口列表。""" open_ports = [] with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as executor: future_map = { executor.submit(tcp_connect_probe, ip, port, timeout): port for port in ports } for future in concurrent.futures.as_completed(future_map): port = future_map[future] if future.result(): open_ports.append(port) return sorted(open_ports)

代码逻辑不复杂：每个端口一个socket，连接成果就是开放；线程池里100个并发端口基本够用，对1000个端口的扫描大概耗时等于端口数除以100再乘以超时时间，所以端口超时直接影响整个扫描时长。参数方面，timeout建议在内网设2秒，跨公网或高延迟链路设3~4秒；workers建议100到200，太高会让本机文件描述符耗尽，报Too many open files错误。

3.2 服务指纹识别：读banner和主动握手，两种策略一结合

拿到开放端口后，服务识别最常见的招数是banner抓取：连上去读返回的文本，比如SSH服务会回SSH-2.0-OpenSSH_7.4，HTTP服务会回Server: nginx/1.18.0。banner抓取准确率还行，但对HTTP这类服务，光读banner拿不到最关键的版本号，因为Nginx会把版本号放在Server头的可选位置，而Apache又经常重写。

更好的做法是主动握手：针对已知端口协议发一段“探针”。比如对HTTP，发一个GET / HTTP/1.1\r\nHost: {ip}\r\n\r\n，然后解析响应头；对SSH，连上去直接读前几字节；对MySQL，先读握手包里的版本字段。我做服务识别时把这两种策略串成一条链：先连上去读3秒内的banner，如果banner文本匹配不到已知特征，再按端口默认协议发送探针。

import socket import re def grab_banner(ip: str, port: int, timeout: float = 3.0) -> str: """读取原始banner，用于服务识别。""" try: sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(timeout) sock.connect((ip, port)) sock.sendall(b"\r\n") # 很多服务收到空命令会回banner banner = sock.recv(4096) return banner.decode("utf-8", errors="replace").strip() except Exception: return "" finally: sock.close() def identify_service(ip: str, port: int) -> str: banner = grab_banner(ip, port) signatures = { r"SSH-\d\.\d-": "ssh", r"^220 .*FTP": "ftp", r"^HTTP/1\.[01]": "http", r"^\*\* Welcome to MySQL": "mysql", } for pattern, service_name in signatures.items(): if re.search(pattern, banner, re.IGNORECASE): return service_name # 命中不了的，再按端口猜一个常见服务作为兜底 default_map = {22: "ssh", 80: "http", 443: "https", 3306: "mysql", 3389: "rdp"} return default_map.get(port, "unknown")

注意grab_banner里sock.sendall(b"\r\n")这行千万不能漏，很多服务（比如纯SMTP、FTP）要等收到回车换行才会回banner，直接recv会卡到超时。identify_service里的正则匹配顺序也很重要：^220 .*FTP必须放在HTTP前面，因为FTP服务器可能也支持HTTP代理，banner长得像HTTP响应，先匹配HTTP就误判了。这个坑我踩过两次，后来把顺序按协议栈的优先级固定下来。

3.3 并发数、超时和重试的取值：这组参数决定扫描是“分钟级”还是“小时级”

很多人第一次跑全端口扫描，看到几万个端口就慌了，然后把超时设成10秒、重试3次，结果一个IP扫了一晚上。我这里给出两组实测下来比较稳的参数组合。

内网全端口扫：超时1.5秒，重试1次，线程并发50~100。因为内网延迟低、丢包率小，超时设置短一点问题不大。公网指定服务扫：超时3秒，重试1次，并发20~50，同时每个IP的扫描总数限制在5个高概率端口（22、80、443、8080、3306）。如果扫的目标是公网资产，还要在发包速率上做平滑，避免触发边界设备的流量阈值。

至于半开扫描，它需要构造原始SYN包，Python里最通用的是Scapy，但Scapy抓包在高并发下性能一般，最终还是会调Nmap。我这里给一个Scapy实现的TCP SYN半开探测的参考写法，用来扫描少量目标没问题，批量扫描就别指望它了：

from scapy.all import IP, TCP, sr1, conf conf.verb = 0 # 关掉Scapy的日志输出 def syn_probe(ip: str, port: int, timeout: float = 2.0) -> bool: """SYN半开扫描：发SYN，收SYN-ACK则端口开放。""" pkt = IP(dst=ip) / TCP(dport=port, flags="S") reply = sr1(pkt, timeout=timeout, verbose=False) if reply is None: return False return reply.haslayer(TCP) and reply.getlayer(TCP).flags & 0x12 == 0x12

flags & 0x12 == 0x12判断是不是SYN-ACK（0x12即SYN和ACK两位都是1）。收到RST包则端口关闭，收不到任何回应多半是防火墙静默丢弃，这种端口会被标记为“filtered”，如果不做处理就会在结果里漏掉，后面避坑章节我会具体讲它怎么坑人。

4. 漏洞与弱口令检测：让扫描结果从“资产清单”变成“风险清单”

前面章节产出的Target对象已经带了IP、端口、服务和版本，这一章要做的就是让结果里有“风险”。漏洞检测和弱口令检测是综合扫描工具最“提气”的部分，也是代码量增长最快的地方。这里的核心逻辑是：先用已识别的版本号去匹配漏洞库，再对特定服务（FTP、SSH、MySQL、Redis）做弱口令验证。需要反复强调的是，所有模块只应在授权范围内运行，这是工具最基本的底线。

4.1 基于版本号的CVE匹配：一个轻量本地指纹库怎么设计

CVE匹配不一定要联网调API。做内网自查时，我习惯维护一个本地JSON或SQLite表，记录服务名称、版本范围与CVE的对应关系。比如openssh版本小于7.4时匹配CVE-2016-6210，nginx版本小于1.18.0时匹配某些已公开的HTTP/2漏洞。这个库不需要覆盖全部CVE，重点是覆盖你用得到的常用中间件，否则匹配率低到没法用。

import json from semver import Version def load_vuln_db(path: str) -> list: with open(path, "r", encoding="utf-8") as f: return json.load(f) def match_cve(service: str, version_str: str, vuln_db: list, ip: str, port: int) -> None: """ 版本匹配逻辑： 遍历漏洞库，service相同、版本号落在[min_version, max_version]区间内才报CVE。 """ for entry in vuln_db: if entry["service"].lower() != service.lower(): continue if version_str == "unknown": continue ver = Version.parse(version_str) if Version.parse(entry["min_version"]) <= ver <= Version.parse(entry["max_version"]): print(f"[!] {ip}:{port} 命中 {entry['cve_id']} 影响版本区间")

semver是Python的语义化版本解析库，能帮你搞定1.8.0和1.8.0-beta之类的比较细节。如果不想引第三方依赖，就用字符串粗暴比较版本号的前两段，但在碰到1.9和1.10这类边界时字符串比较会出错，所以我还是推荐上semver。这里有一个容易被忽略的现实问题：grab_banner拿到的版本字符串经常是OpenSSH_7.4p1 Debian 10+deb9u1这种带后缀的，直接Version.parse会抛异常，所以要先正则抽取数字点号部分。

4.2 弱口令检测：不是暴力破解，是用低频率、按策略验证

弱口令检测最忌讳的是把它做成爆破脚本，高频尝试会锁账号、触发风控，甚至把目标服务打挂。我这里的做法是：只针对特定服务和高风险端口做一次“策略性验证”，密码字典控制在50个以内，且每个目标之间间隔至少1秒。

import paramiko username_list = ["root", "admin", "test"] password_list = ["123456", "admin", "root", "password", "toor", "test123"] def ssh_check(ip: str, port: int, username: str, password: str, timeout: float = 5.0) -> bool: """单组账号密码的SSH弱口令验证，失败异常静默。""" client = paramiko.SSHClient() client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) try: client.connect(ip, port=port, username=username, password=password, timeout=timeout) return True except Exception: return False finally: client.close() def ssh_weak_password_check(ip: str, port: int) -> list: weak_list = [] for user in username_list: for pwd in password_list: if ssh_check(ip, port, user, pwd): weak_list.append(f"{user}:{pwd}") break # 同一个用户命中一组后就停止，减少尝试次数 return weak_list

break这行值得多说一句：同一个用户名只要命中一组弱口令就够了，没必要把50个密码全部试完；这既降低了目标被锁定的概率，也缩短了扫描时间。要警惕的是，paramiko首次连接时会对陌生主机的host key做校验，上面代码里设置了AutoAddPolicy自动接受，这在授权扫描场景能免掉交互，但在生产环境如果代码被改去连非授权目标，这种行为是有风险的，所以这个参数放在这里的同时，必须在配置文件里标明“仅限授权测试”。

其他服务的弱口令检测思路类似，MySQL用pymysql、Redis用redis-py或直接socket发AUTH命令、FTP用ftplib，注意各个库的包名和连接超时参数不完全一致，但核心逻辑就是“连接建立成功即命中”。Redis弱口令检测的坑在于，Redis默认不允许远程认证，很多配置是启动后只监听127.0.0.1，所以探测前要先看端口是否暴露在外网或跨网段。

4.3 Web层面的信息收集：目录扫描和HTTP指纹别漏掉

综合扫描工具不能不管Web服务。这里我加了两个轻量模块：HTTP响应头指纹识别和常见目录扫描。响应头识别很简单，就是发GET请求后解析Server、X-Powered-By、Set-Cookie等字段；目录扫描则基于一个常见的目录字典，对每个路径发HEAD请求，看状态码是200还是403还是404。目录字典控制在500条左右，太大了会拖慢整个扫描流程。

import requests dir_list = ["admin", "login", "api", "backup", ".git", "uploads", "phpmyadmin", "config", "test"] def http_probe(ip: str, port: int, path: str, timeout: float = 3.0) -> tuple: """返回状态码和Server头，用于判断是否存在敏感路径。""" url = f"http://{ip}:{port}{path}" if port != 443 else f"https://{ip}:{port}{path}" try: resp = requests.get(url, timeout=timeout, allow_redirects=False) return resp.status_code, resp.headers.get("Server", "") except requests.RequestException: return 0, "" status_403_list = [] for path in dir_list: code, server = http_probe(ip, port, f"/{path}/") if code in (200, 301): print(f"[+] {path} 可访问，Server: {server}") elif code == 403: status_403_list.append(path)

allow_redirects=False是为了避免目录扫描跟随30x跳转到登录页造成误报，很多应用会把/admin302到/login，如果跟随下去结果全是200，反而掩盖了真实状态。另外路径末尾要加斜杠，有些Nginx配置会对无斜杠的目录返回403而非301，加了斜杠才能测到真实目录行为。这里有个小技巧：对403的路径单独收录，因为很多敏感目录会挡掉匿名访问但返回403，属于“存在但被保护”，也要写进报告。

4.4 扫描结果的归并去重：同一个漏洞别在报告里出现三次

综合工具最容易出现的问题就是这个：同一个目标同一个端口，被TCP探测标记为开放、服务识别标记为ssh、verison命中CVE、弱口令模块也命中，如果每层都往Target对象里追加记录而不去重，报告就会同一个漏洞出现三次。我在各检测模块完成之后统一跑一个去重函数，只按cve_id + ip + port做去重，顺便把相同cve_id的多个端口合并显示。

def deduplicate_results(targets: list) -> list: seen = set() for t in targets: unique_cves = [] for cve in t.cve_list: key = (t.ip, t.port, cve) if key not in seen: seen.add(key) unique_cves.append(cve) t.cve_list = unique_cves return targets

去重逻辑虽然简单，但唯一键的选择很重要。我见过有人把service也加进唯一键里，结果同一个CVE在SSH和FTP上出现两次被当成两个漏洞；还有人不用IP+端口做键，直接把所有CPE匹配出来的CVE全列进cve_list，报告里密密麻麻全是重复项。去重的本质问题不只是数据整洁，而是影响后面风险评分的准确性，同一个CVE重复上报会让业务方对风险等级产生误判。

5. 综合扫描工具常见问题排查：这5个坑我建议你直接背下来

5.1 扫描器启动即报Address already in use

现象：程序启动后立刻抛OSError: [Errno 98] Address already in use，或者服务识别模块连接下一个端口时报错。

原因：多半是你用了socket.socket()创建探测连接时设了SO_REUSEADDR，但在并发场景下两个线程同时bind到了同一个本地端口。我这里处理的办法是让socket只connect不bind，让系统自动分配临时端口；如果确实需要固定源端口（比如要过防火墙白名单），就把并发数降下来，并用端口范围池做分配。

def create_tcp_socket(timeout: float = 2.0) -> socket.socket: sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(timeout) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) return sock

解决方式：SO_REUSEADDR只影响bind不监听时的行为，并不会导致Address already in use；真正要查的是有没有并发地监听。如果代码里没有主动bind，那就去查系统里是不是已经有别的服务占了你要用的端口，用lsof -i :端口排查一下。

5.2 扫描结果全是closed，但手工telnet端口是通的

现象：端口扫描器跑完一个开着的端口却报closed，手工telnet ip port却通。

原因：这是老坑了。目标主机上装了防火墙（iptables/nftables/安全组），对无payload的连接（空banner的TCP流）默认丢弃或拒绝，而手工telnet往往带上了终端初始化命令，所以能通。还有一个可能是扫描线程太多触发了目标的连接频率限制。

解决：对这类目标改用带payload的探测，比如先发一个空HTTP请求再读响应；同时把并发降到50以下，单个端口连续重试不超过2次。如果这两个都不行，考虑是不是目标安全设备把同IP的高频连接识别成扫描，做了黑名单，这种要做源IP轮换或放慢速率。

5.3 服务识别把SSH识别成HTTP

现象：目标开放22端口，banner明明是SSH-2.0-OpenSSH_7.4，但服务识别模块返回的是http。

原因：正则匹配优先级顺序错了。我在3.2节里提过，如果HTTP的正则^HTTP/1\.[01]写在SSH的^SSH前面，而banner刚好开头以“HTTP”开头（比如某些SSH服务开启了代理模式返回了HTTP响应），就会误判。还有一种情况是banner抓取不完整，只抓到一行HTTP/1.1 400 Bad Request，被误认为了Web服务。

解决：跟踪识别顺序，把协议栈的确定特征放前面；SSH的特征SSH-\d比HTTP的^HTTP更可靠，应当先匹配。同时抓banner时用recv(4096)多读一些，别只读第一行。

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

现象：扫描完半小时后，业务方投诉某个管理后台登录不进去，一看日志是扫描器高频尝试登录。

原因：这是最严重的坑。有些服务的账号锁定策略是连续失败5次锁定15分钟，弱口令检测默认字典即使只有50条，对5个用户跑下来也会触发锁号。我之前遇到过，密码字典里第6条碰对了，但前面5条已经触发了锁定策略，后续正确账号也被锁住。

解决：给弱口令模块加双层保险。第一层是全局限速，同一个目标两个弱口令探测之间至少间隔2秒；第二层是对账号做状态记忆，同一个用户名只要失败3次就跳过该用户名，不再尝试更多密码。实践中还要在配置文件里允许使用者指定“最小延迟”和“失败次数阈值”，防止不同环境的锁定策略差异。

class WeakPassChecker: def __init__(self, max_failures_per_user: int = 3, min_interval: float = 2.0): self.max_failures = max_failures_per_user self.interval = min_interval self.fail_count = {} self.last_try = {} def can_try(self, user: str) -> bool: now = time.time() if self.fail_count.get(user, 0) >= self.max_failures: return False if now - self.last_try.get(user, 0) < self.interval: return False return True

5.5 扫描结果没入库，服务一重启全丢了

现象：扫描完生成了一份JSON，第二天机器重启后要重新生成报告，发现数据没了。

原因：综合工具如果只在内存里保存Target列表，任何宕机、误操作kill都会导致数据丢失。Python脚本里如果没做持久化，Ctrl+C后连结束信息都来不及写。

解决：在探测调度层加一个“边扫边写”的机制，扫描进度每完成1%就向SQLite写入一次结果。这样即使中途被杀，已经完成的端口状态也不会白扫。SQLite文件本地存储，不需要装MySQL，足够单机扫描的场景用。

import sqlite3 def init_db(path: str) -> None: conn = sqlite3.connect(path) conn.execute(""" CREATE TABLE IF NOT EXISTS scan_result ( id INTEGER PRIMARY KEY AUTOINCREMENT, ip TEXT NOT NULL, port INTEGER NOT NULL, service TEXT, version TEXT, status TEXT, cve_list TEXT, weak_passwords TEXT, scan_time TEXT DEFAULT (datetime('now','localtime')) ) """) conn.commit() conn.close()

写库时注意cve_list和weak_passwords是列表类型，SQLite没有数组字段，建议用JSON序列化后再存，查询时再解析。这个表结构还能直接喂给后面的报告模块，不用再回扫一次。

6. 进阶思路：结果推送与定时巡检，把扫描器变成常态化工具

综合扫描工具真正产生价值的地方不在“扫一次出报告”，而在于把扫描能力嵌进日常的巡检流程里。我自己常用的进阶做法是：扫描完成后把新增风险项推送进企业微信群或飞书群，高危漏洞单独标红；同时在crontab里设定每周日凌晨2点自动跑一次内网资产巡检，跑完自动生成HTML报告放到固定路径。这样运维和合规同事不用手动触发扫描器，风险变化趋势也能追溯。

定时巡检的实现不复杂，核心是扫描入口函数必须支持命令行参数输入IP段和输出目录。在__main__里把参数解析、调度执行、结果入库三段串起来，然后crontab里写一行0 2 * * 0 cd /opt/scan && python3 main.py --target 192.168.0.0/16 --out /data/reports就行。需要注意两个细节：一是扫描期间要防止两个任务重叠，可以在入口文件加一个lockfile，启动时检查文件锁存在就直接退出；二是报告文件名带日期，比如report_20250101.html，方便追溯某一次扫描的问题。

关于风险评分，我给扫描结果加了一个简单的加权算法：每个漏洞按其CVSS严重程度打分，CVE-2017严重级别9.8，弱口令按服务重要性加权（SSH的弱口令权重比FTP高），再按IP聚合成单资产风险分。这个分不用很复杂，目的只是让报告首页能按风险从高到低排个序，让工程师先处理高危资产。这一步是很多人在“扫描工具”项目里忽略的，但我认为它才是工具落地的最后一块拼图——没有排序的风险清单，就是一份没法行动的黑匣子。

最后说一个我自己的个人教训：这套工具的每一行代码里，最容易让人放松警惕的是重试逻辑和超时参数。我早期调公网扫描时把超时设成5秒、重试设成3次，结果一次扫描跑了22个小时还没完，后来把所有探测统一收敛到超时2.5秒、重试1次，加上带速率的并发调度，整体扫描时间缩短了40倍。参数这个东西，不是越大越稳，而是和目标网络的实际延迟匹配才稳。希望这些思路和源码能帮你少走几个坑，把网络安全扫描工具做成真正能在团队里落地运转的常态化设施。

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