Python构建可审计的综合网络安全扫描器

简介：这是一套基于Python3开发的综合网络安全扫描工具源码，面向安全工程师、渗透测试人员及网络安全学习者，旨在提供一套开箱即用、功能完备的授权安全评估解决方案。工具覆盖敏感文件探测、WAF/CDN识别、端口与服务识别、操作系统指纹分析、弱口令检测、常见漏洞利用（如Struts、WebLogic、Redis未授权访问等）及旁站查询等核心能力，兼顾甲方自测与乙方授权场景，强调合法合规使用。资源包为6.98MB的ZIP压缩包，共41个文件：其中31个Python脚本构成主功能模块（如vuln.py、waf.py、osdetect.py、各类CVE利用脚本），2个JSON配置文件（apps.json等）支撑规则与应用识别，2个JPG图标文件用于界面标识，另含LICENSE、README、requirements.txt及GeoLite2-ASN.mmdb等关键支持文件，目录结构清晰，模块职责分明。已有333人学习下载，可直接部署运行，快速掌握实战级扫描工具的设计逻辑、插件化架构与漏洞检测实现细节。

1. 为什么一个“基于Python的综合网络安全扫描工具”不能只靠nmap+requests拼凑完事？

你手头有个需求：写个能跑在内网、不依赖云API、不调用商业引擎、纯本地执行的扫描工具——要能识别开放端口、探测服务指纹、检查常见Web路径、抓取SSL证书信息、甚至对HTTP响应做基础安全头分析。这时候搜“Python 网络安全 扫描工具”，满屏是python-nmap封装、socket连端口、urllib发GET的碎片脚本。但真把它们塞进一个main.py里跑起来，你会发现：超时控制乱套、并发一开就卡死、HTTPS跳转后证书链验不了、403页面和真实不存在的路径根本分不清、扫描结果散落在5个字典里没法导出……这不是功能缺失，是工程断层：缺少统一任务调度、状态追踪、结果归一化和错误隔离机制。这个标题不是教你怎么写socket.connect()，而是讲清楚——如何用Python构建一个可中断、可复用、可审计、可横向扩展的扫描底座。适合刚脱离CTF靶场、正接手企业内网资产普查的渗透测试新人，也适合想把零散PoC整合进自动化流程的安全开发工程师。它不承诺发现0day，但能让你每次扫描都留下可回溯的操作日志、结构化输出和明确的失败归因。

2. 从零搭建扫描器骨架：核心模块划分与依赖选型逻辑

一个真正“综合”的扫描工具，绝不是把nmap、httpx、sslscan命令行包装一遍。它必须解决三个底层矛盾：异构协议统一调度（TCP/HTTP/HTTPS/SSL）、资源竞争可控收敛（避免打爆目标或本机）、结果语义一致化（让端口、路径、证书、头信息能被同一规则引擎消费）。我们不用重造轮子，但必须知道每个轮子为什么选它。

2.1 模块分层设计：为什么必须拆成core/scanner/output三层？

• core/：负责任务队列、线程/协程池管理、全局配置加载、信号捕获（Ctrl+C中断）、日志分级（DEBUG/INFO/WARN/ERROR）。这里不用asyncio硬上，而用concurrent.futures.ThreadPoolExecutor+queue.Queue组合——因为多数扫描动作（如Socket连接、HTTP请求）本质是I/O阻塞，线程比协程更易调试、更少隐式状态。
• scanner/：按协议垂直切分。port_scanner.py（TCP SYN半开不可行，改用connect超时检测）、http_scanner.py（带重定向跟随、User-Agent轮换、基础头注入）、ssl_scanner.py（用ssl原生库解析证书而非调openssl命令）、dir_scanner.py（支持字典爆破+状态码过滤+内容长度去重）。每个子模块只暴露scan(target: str, options: dict) -> ScanResult接口，返回统一结构体。
• output/：不直接print，而是通过OutputManager抽象类实现to_json()/to_csv()/to_html()。关键点在于：所有模块产出的ScanResult必须继承自BaseResult，强制定义target,timestamp,module,status,data五字段——这是后续做关联分析（比如“80端口开放且返回了X-Powered-By: PHP/7.4”）的基础。

提示：别急着写代码。先画一张模块通信图：core.scheduler读取config.yaml生成任务列表 → 分发给scanner.*实例 → 结果经core.result_collector聚合 → 交由output.*格式化。这张图决定了你后续90%的调试成本。

2.2 关键依赖选型：为什么弃用python-nmap而用python-libnmap？为什么httpx不如requests+urllib3可控？

安装命令必须带约束：

pip install "requests>=2.28.0,<2.30.0" "urllib3>=1.26.12,<1.27.0" "pyOpenSSL>=23.0.0,<24.0.0" "python-libnmap>=0.7.0,<0.8.0"

注意：requests 2.30.0+移除了urllib3.util.retry.Retry的raise_on_status=False参数，会导致重试逻辑失效；pyOpenSSL 24.0.0升级了cryptography依赖，与旧版paramiko冲突——这些版本锁死不是保守，是血泪经验。

3. 端口扫描模块实现：如何让TCP连接既快又准，避开防火墙干扰？

端口扫描是整个工具的入口，它的准确率直接决定后续模块是否启动。很多人用socket.connect()加settimeout(1)，结果在扫描192.168.1.0/24时，80%的IP返回Connection refused（目标关机）和Timeout（防火墙丢包）混在一起，根本无法区分。我们必须引入三次探测+状态交叉验证机制。

3.1 基于socket的精准连接检测：三次探测法代码实现

import socket import time from typing import Tuple, Optional def probe_port(host: str, port: int, timeout: float = 2.0) -> Tuple[bool, str, Optional[str]]: """ 对单个端口执行三次探测，返回 (is_open, reason, banner) reason: 'open' / 'closed' / 'filtered' / 'unreachable' banner: 仅当open且能recv时返回前128字节 """ # 第一次：快速connect，超时1秒 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(1.0) try: sock.connect((host, port)) # 连接成功，尝试读取banner（最多128字节，不阻塞） sock.settimeout(0.5) try: banner = sock.recv(128).decode('utf-8', errors='ignore') except (socket.timeout, OSError): banner = None sock.close() return True, 'open', banner except socket.timeout: # 第二次：延长超时至3秒，排除网络抖动 sock.settimeout(3.0) try: sock.connect((host, port)) sock.close() return True, 'open', None except socket.timeout: # 第三次：发送SYN后立即关闭，看是否收到RST（需root权限，此处降级为ICMP探测） try: # 使用系统ping判断主机是否存活（规避ICMP禁用场景） import subprocess result = subprocess.run(['ping', '-c', '1', '-W', '1', host], capture_output=True, text=True) if result.returncode == 0: return False, 'filtered', None else: return False, 'unreachable', None except Exception: return False, 'filtered', None except ConnectionRefusedError: sock.close() return False, 'closed', None except Exception as e: sock.close() return False, f'error:{str(e)[:20]}', None except ConnectionRefusedError: sock.close() return False, 'closed', None except Exception as e: sock.close() return False, f'error:{str(e)[:20]}', None

逻辑说明：

• 第一次1秒超时是主力判断，覆盖90%正常网络；
• 第二次3秒超时专治高延迟链路（如跨省专线）；
• 第三次用ping辅助判断：若ping不通，大概率是主机下线（unreachable）；若ping通但端口连不上，基本确定是防火墙拦截（filtered）。
• banner只在第一次连接成功时尝试获取，避免对慢服务造成额外延迟。

参数说明：

• timeout=2.0是总耗时上限，实际执行中三次探测最大耗时约5秒（1+3+1），远低于传统单次5秒扫描；
• banner解码用errors='ignore'防止二进制数据崩掉，生产环境建议用chardet自动识别编码；
• ping命令用-W 1指定等待1秒，避免在丢包率高的网络中卡死。

3.2 并发控制与结果聚合：如何避免线程数设错导致扫描器变DoS工具？

盲目开100线程扫C段？轻则触发目标IDS告警，重则本机ulimit -n耗尽（Linux默认1024文件描述符）。必须实现动态并发窗口：

from concurrent.futures import ThreadPoolExecutor, as_completed import threading class PortScanner: def __init__(self, max_workers: int = 20): self.max_workers = max_workers self._lock = threading.Lock() self.results = [] def scan_range(self, host: str, port_range: range) -> list: # 动态调整worker数：每10个端口预留1个worker，上限max_workers effective_workers = min(self.max_workers, len(port_range) // 10 + 1) with ThreadPoolExecutor(max_workers=effective_workers) as executor: # 提交所有任务 future_to_port = { executor.submit(probe_port, host, port): port for port in port_range } # 收集结果，带进度提示 for future in as_completed(future_to_port): port = future_to_port[future] try: is_open, reason, banner = future.result() with self._lock: self.results.append({ 'host': host, 'port': port, 'is_open': is_open, 'reason': reason, 'banner': banner, 'timestamp': time.time() }) except Exception as exc: with self._lock: self.results.append({ 'host': host, 'port': port, 'is_open': False, 'reason': f'exception:{str(exc)[:30]}', 'banner': None, 'timestamp': time.time() }) return self.results

关键设计点：

• effective_workers根据端口数量动态计算，避免小范围扫描（如只扫22,80,443）时开太多线程；
• as_completed()保证结果按完成顺序返回，配合threading.Lock()写入共享列表，比executor.map()更易捕获单个异常；
• 每个结果强制包含timestamp，为后续做时间序列分析（如“某IP在10分钟内连续触发3次filtered”）留接口。

4. Web服务深度探测：从HTTP头分析到SSL证书链验证

端口开着只是开始。80/443开放后，必须回答：这是Nginx还是Apache？启用了HSTS吗？证书是否过期？有没有暴露敏感路径？这一层决定扫描器能否从“端口列表器”升级为“风险定位器”。

4.1 HTTP响应头安全分析：6个必检Header及其业务含义

不要只打印response.headers。以下6个Header必须结构化解析并打分：

def analyze_headers(response) -> dict: issues = [] headers = {k.lower(): v for k, v in response.headers.items()} # Server头检查 if 'server' in headers: server_match = re.search(r'(nginx|apache|iis)/([\d.]+)', headers['server'], re.I) if server_match: product, version = server_match.groups() # 此处应调用本地CVE数据库查询，简化为伪代码 if is_version_vulnerable(product, version): issues.append(f"{product} {version} has known CVEs") # HSTS检查 if 'strict-transport-security' not in headers: issues.append("Missing HSTS header") else: hsts = headers['strict-transport-security'] if 'max-age=' not in hsts or int(re.search(r'max-age=(\d+)', hsts).group(1)) < 31536000: issues.append("HSTS max-age too short (<1 year)") # 其他Header同理... return {"issues": issues, "headers": dict(response.headers)}

注意：is_version_vulnerable()函数必须对接本地CVE数据库（如NVD JSON feed缓存），不能实时调API——否则扫描100个域名会触发速率限制。我一般用sqlite3建表cve_db，字段product TEXT, version TEXT, cve_id TEXT, severity TEXT，每日凌晨用curl拉取增量更新。

4.2 SSL证书链验证：为什么ssl.create_default_context()不够用？

requests.get("https://...", verify=True)只能验证证书是否由可信CA签发，但无法检测：

• 证书是否在notBefore之前生效（时钟不同步导致）；
• subjectAltName是否包含扫描域名（防域名欺骗）；
• OCSP Stapling是否启用（影响吊销状态实时性）；
• 密钥长度是否≥2048位（RSA）或≥256位（ECDSA）。

import ssl from OpenSSL import crypto def check_ssl_cert(hostname: str, port: int = 443) -> dict: try: # 创建上下文，禁用证书验证（我们自己验） context = ssl.create_default_context() context.check_hostname = False context.verify_mode = ssl.CERT_NONE with socket.create_connection((hostname, port), timeout=5) as sock: with context.wrap_socket(sock, server_hostname=hostname) as ssock: # 获取原始证书 cert_der = ssock.getpeercert(binary_form=True) cert = crypto.load_certificate(crypto.FILETYPE_ASN1, cert_der) # 解析基本信息 subject = cert.get_subject() issuer = cert.get_issuer() not_before = cert.get_notBefore().decode() not_after = cert.get_notAfter().decode() serial = hex(cert.get_serial_number()) # 检查域名匹配（subjectAltName优先） san_list = [] for i in range(cert.get_extension_count()): ext = cert.get_extension(i) if b'subjectAltName' in ext.get_short_name(): san_list = [s.strip() for s in str(ext).split(',')] break # 验证域名是否在SAN中 domain_ok = any(hostname in san for san in san_list) if san_list else \ hostname == subject.CN.decode() if hasattr(subject, 'CN') else False return { "valid": domain_ok and (time.time() > ssl.cert_time_to_seconds(not_before)) and \ (time.time() < ssl.cert_time_to_seconds(not_after)), "subject": str(subject), "issuer": str(issuer), "not_before": not_before, "not_after": not_after, "serial": serial, "san": san_list, "key_size": cert.get_pubkey().bits() } except Exception as e: return {"error": str(e)}

参数说明：

• context.check_hostname = False：避免wrap_socket阶段校验失败，我们自己做；
• cert.get_pubkey().bits()直接获取密钥长度，比解析get_subject().get_components()更可靠；
• ssl.cert_time_to_seconds()将ASN.1时间转为Unix时间戳，用于和time.time()比较。

5. 避坑指南：扫描器开发中5个高频翻车点及血泪解法

写扫描工具最痛苦的不是功能实现，而是那些让结果不可信、过程不可控、复现不可靠的细节。以下是我在3个企业内网扫描项目中踩过的坑，按发生频率排序：

5.1 现象：扫描结果里大量filtered状态，但人工nmap -sS确认是open

原因：Pythonsocket.connect()在Linux上默认使用connect()系统调用，而nmap -sS用的是原始套接字发SYN包。当目标开启iptables -A INPUT -p tcp --tcp-flags ALL NONE -j DROP（防空连接扫描）时，connect()会因收不到SYN-ACK而超时，但SYN包本身已被防火墙静默丢弃，导致误判为filtered。
解决：放弃纯Python实现，改用python-libnmap调用系统nmap，并指定-sS参数。在core/scheduler.py中增加判断：

if platform.system() == "Linux" and os.geteuid() == 0: # root用户启用SYN扫描 nmap_args = ["-sS", "-T4", "--min-rate=100"] else: # 普通用户降级为connect扫描 nmap_args = ["-sT", "-T4"]

5.2 现象：扫描HTTPS站点时，requests抛SSLError: certificate verify failed，但浏览器能正常访问

原因：目标使用私有CA证书（如企业内部PKI），而requests只信任Mozilla CA列表，不读取系统证书存储。
解决：不简单设verify=False（会丢失证书链信息），而是将系统证书路径注入：

import ssl import certifi # Linux: /etc/ssl/certs/ca-bundle.crt, macOS: /etc/ssl/cert.pem system_ca = "/etc/ssl/certs/ca-bundle.crt" if os.path.exists("/etc/ssl/certs/ca-bundle.crt") else certifi.where() session = requests.Session() session.verify = system_ca

5.3 现象：多线程扫描时，urllib3抛Max retries exceeded，但单线程正常

原因：urllib3的PoolManager默认连接池大小为10，20个线程共用一个Session时，连接复用竞争导致超时。
解决：为每个线程创建独立Session，并显式设置连接池：

def create_session(): session = requests.Session() adapter = requests.adapters.HTTPAdapter( pool_connections=20, pool_maxsize=20, max_retries=urllib3.util.retry.Retry( total=3, backoff_factor=0.3, status_forcelist=[502, 503, 504], ) ) session.mount('http://', adapter) session.mount('https://', adapter) return session

5.4 现象：扫描结果JSON导出后，datetime对象序列化失败

原因：json.dumps()不支持datetime，而扫描时间戳用datetime.now()生成。
解决：统一用time.time()（float）替代，或自定义JSONEncoder：

class ScanJSONEncoder(json.JSONEncoder): def default(self, obj): if isinstance(obj, datetime): return obj.isoformat() return super().default(obj) json.dumps(results, cls=ScanJSONEncoder)

5.5 现象：dir_scanner爆破目录时，403 Forbidden和404 Not Found返回相同HTML内容，无法区分

原因：某些WAF（如Cloudflare）对403/404返回相同错误页，仅靠状态码和内容长度无法判断。
解决：增加响应指纹比对——采集已知404页面样本（如/nonexistent12345.html），计算其MD5，后续所有响应与之比对：

# 预扫描阶段 sample_404 = requests.get(f"https://{target}/nonexistent_{int(time.time())}.html").content self._404_fingerprint = hashlib.md5(sample_404).hexdigest() # 扫描时 for path in wordlist: r = session.get(f"https://{target}{path}") if r.status_code == 403 and hashlib.md5(r.content).hexdigest() != self._404_fingerprint: # 真实403，非WAF伪装 results.append({"path": path, "status": 403})

6. 让扫描结果真正可用：构建可关联的风险视图与自动化处置链

扫描器的价值不在发现多少端口，而在把离散数据变成决策依据。比如：发现192.168.1.10:80开放，Server: Apache/2.4.29，X-Powered-By: PHP/7.2.24，SSL证书过期——这四个事实单独看是信息，组合起来就是一条高危告警：“Apache 2.4.29 + PHP 7.2.24 组合存在CVE-2019-0211（privilege escalation），且证书过期导致TLS降级风险”。这才是综合扫描器该干的事。

6.1 风险关联引擎：用YAML规则定义“事实组合即风险”

在rules/目录下放web_risk_rules.yaml：

- id: "apache_php_cve2019_0211" name: "Apache + PHP privilege escalation" description: "Apache 2.4.29-2.4.39 with PHP 7.2.0-7.2.24 allows privilege escalation via mod_proxy" condition: - module: "http_scanner" field: "headers.Server" op: "regex" value: "Apache/2\\.4\\.(29|3[0-9])" - module: "http_scanner" field: "headers.X-Powered-By" op: "regex" value: "PHP/7\\.2\\.(0|[1-9]|[1-9][0-9]|2[0-4])" severity: "high" remediation: "Upgrade Apache to 2.4.41+, PHP to 7.2.25+" - id: "ssl_expired" name: "SSL certificate expired" condition: - module: "ssl_scanner" field: "valid" op: "eq" value: false severity: "medium" remediation: "Renew certificate and check system clock"

6.2 规则引擎执行器：如何用最少代码实现条件匹配？

import yaml import re from typing import List, Dict, Any class RiskEngine: def __init__(self, rules_path: str): with open(rules_path) as f: self.rules = yaml.safe_load(f) def match_rules(self, scan_results: List[Dict]) -> List[Dict]: alerts = [] for rule in self.rules: # 将rule.condition中每个条件映射到对应scan_result matched = True for cond in rule["condition"]: # 找到module匹配的结果 target_result = next((r for r in scan_results if r.get("module") == cond["module"]), None) if not target_result: matched = False break # 提取field值（支持嵌套，如headers.Server） field_parts = cond["field"].split(".") value = target_result for part in field_parts: if isinstance(value, dict) and part in value: value = value[part] else: matched = False break if not matched: break # 执行比较 if cond["op"] == "regex": if not re.search(cond["value"], str(value)): matched = False elif cond["op"] == "eq": if str(value) != str(cond["value"]): matched = False if matched: alerts.append({ "rule_id": rule["id"], "name": rule["name"], "severity": rule["severity"], "remediation": rule["remediation"], "matched_results": [r for r in scan_results if r.get("module") in [c["module"] for c in rule["condition"]]] }) return alerts # 使用示例 engine = RiskEngine("rules/web_risk_rules.yaml") alerts = engine.match_rules(all_scan_results)

关键设计：

• field: "headers.Server"支持点号嵌套，适配任意层级字典；
• op: "regex"和op: "eq"覆盖90%匹配场景，无需引入JMESPath等重型表达式；
• matched_results保留原始数据引用，方便前端展示上下文（如点击告警直接跳转到对应的HTTP响应头）。

6.3 自动化处置链：从告警到工单的最小闭环

很多团队卡在“发现风险→人工填工单→等修复→再扫描”循环里。我们可以用扫描器自身打通最后一公里：

def create_jira_ticket(alert: dict, jira_config: dict): """向Jira创建缺陷工单，返回ticket_id""" url = f"{jira_config['base_url']}/rest/api/3/issue" auth = (jira_config["user"], jira_config["api_token"]) payload = { "fields": { "project": {"key": jira_config["project_key"]}, "summary": f"[SCAN] {alert['name']}", "description": f"Severity: {alert['severity']}\n\n" f"Remediation: {alert['remediation']}\n\n" f"Affected targets:\n" + "\n".join(set(r["target"] for r in alert["matched_results"])), "issuetype": {"name": "Bug"} } } resp = requests.post(url, json=payload, auth=auth) if resp.status_code == 201: return resp.json()["key"] else: logger.error(f"Jira ticket creation failed: {resp.text}") return None # 在main.py中调用 if alerts: jira_config = load_config("jira.yaml") for alert in alerts: ticket_id = create_jira_ticket(alert, jira_config) if ticket_id: print(f"✅ Created Jira ticket {ticket_id} for {alert['name']}")

我的习惯是：扫描器只负责创建工单，不负责关闭。因为修复验证必须由人工或独立的验收脚本完成——这是安全红线。曾经有团队让扫描器自动关闭工单，结果因WAF规则变更导致误报，工单关闭后漏洞实际仍在。现在我的扫描报告末尾永远有一行加粗提示：“所有工单需经人工复核后关闭”。

希望帮到你。

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