尧图网站建设 
简介这是一套基于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的综合网络安全扫描工具”不能只靠nmaprequests拼凑完事你手头有个需求写个能跑在内网、不依赖云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/负责任务队列、线程/协程池管理、全局配置加载、信号捕获CtrlC中断、日志分级DEBUG/INFO/WARN/ERROR。这里不用asyncio硬上而用concurrent.futures.ThreadPoolExecutorqueue.Queue组合——因为多数扫描动作如Socket连接、HTTP请求本质是I/O阻塞线程比协程更易调试、更少隐式状态。scanner/按协议垂直切分。port_scanner.pyTCP 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不如requestsurllib3可控依赖选用理由替代方案踩坑记录python-libnmap解析nmap XML输出稳定支持增量解析大扫描不爆内存API返回对象而非字符串python-nmap底层调subprocess.Popen易被杀进程导致XML残缺os.system(nmap -oX ...)XML未闭合标签导致xml.etree.ElementTree解析崩溃无重试机制requestsurllib3.util.retry.Retry可精细控制重试策略如对503状态码重试3次对401不重试Session复用连接池verifyFalse时仍能获取原始证书链httpx默认启用HTTP/2某些老旧Web服务器如Apache 2.2返回400 Bad Request且无明确报错排查耗时3小时pyOpenSSL直接解析PEM证书提取notBefore/notAfter/subjectAltName等字段比ssl标准库更易处理自签名证书异常ssl.get_server_certificate()无法获取OCSP stapling状态且对SNI缺失的服务器会握手失败安装命令必须带约束pip install requests2.28.0,2.30.0 urllib31.26.12,1.27.0 pyOpenSSL23.0.0,24.0.0 python-libnmap0.7.0,0.8.0注意requests 2.30.0移除了urllib3.util.retry.Retry的raise_on_statusFalse参数会导致重试逻辑失效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, errorsignore) 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_outputTrue, textTrue) 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, ferror:{str(e)[:20]}, None except ConnectionRefusedError: sock.close() return False, closed, None except Exception as e: sock.close() return False, ferror:{str(e)[:20]}, None逻辑说明第一次1秒超时是主力判断覆盖90%正常网络第二次3秒超时专治高延迟链路如跨省专线第三次用ping辅助判断若ping不通大概率是主机下线unreachable若ping通但端口连不上基本确定是防火墙拦截filtered。banner只在第一次连接成功时尝试获取避免对慢服务造成额外延迟。参数说明timeout2.0是总耗时上限实际执行中三次探测最大耗时约5秒131远低于传统单次5秒扫描banner解码用errorsignore防止二进制数据崩掉生产环境建议用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_workerseffective_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: fexception:{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必须结构化解析并打分Header检查逻辑风险等级业务含义Server正则匹配nginx/[0-9.]、Apache/[0-9.]版本号查CVE数据库高暴露精确中间件版本降低漏洞利用门槛X-Powered-By存在即标记值为PHP/7.4.33等具体版本中同上且常被用于指纹识别Strict-Transport-Security检查max-age是否≥315360001年有includeSubDomains低缺失为中危缺失意味着HTTP→HTTPS降级攻击可行Content-Security-Policy是否存在且default-src不为none或*中CSP缺失或配置过宽XSS风险上升X-Frame-Options值为DENY或SAMEORIGIN低缺失为中危缺失导致点击劫持ClickjackingX-Content-Type-Options值必须为nosniff低缺失为中危缺失可能触发MIME类型嗅探导致JS/CSS被误执行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(rmax-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://..., verifyTrue)只能验证证书是否由可信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), timeout5) as sock: with context.wrap_socket(sock, server_hostnamehostname) as ssock: # 获取原始证书 cert_der ssock.getpeercert(binary_formTrue) 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 bsubjectAltName 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-rate100] else: # 普通用户降级为connect扫描 nmap_args [-sT, -T4]5.2 现象扫描HTTPS站点时requests抛SSLError: certificate verify failed但浏览器能正常访问原因目标使用私有CA证书如企业内部PKI而requests只信任Mozilla CA列表不读取系统证书存储。解决不简单设verifyFalse会丢失证书链信息而是将系统证书路径注入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_ca5.3 现象多线程扫描时urllib3抛Max retries exceeded但单线程正常原因urllib3的PoolManager默认连接池大小为1020个线程共用一个Session时连接复用竞争导致超时。解决为每个线程创建独立Session并显式设置连接池def create_session(): session requests.Session() adapter requests.adapters.HTTPAdapter( pool_connections20, pool_maxsize20, max_retriesurllib3.util.retry.Retry( total3, backoff_factor0.3, status_forcelist[502, 503, 504], ) ) session.mount(http://, adapter) session.mount(https://, adapter) return session5.4 现象扫描结果JSON导出后datetime对象序列化失败原因json.dumps()不支持datetime而扫描时间戳用datetime.now()生成。解决统一用time.time()float替代或自定义JSONEncoderclass ScanJSONEncoder(json.JSONEncoder): def default(self, obj): if isinstance(obj, datetime): return obj.isoformat() return super().default(obj) json.dumps(results, clsScanJSONEncoder)5.5 现象dir_scanner爆破目录时403 Forbidden和404 Not Found返回相同HTML内容无法区分原因某些WAF如Cloudflare对403/404返回相同错误页仅靠状态码和内容长度无法判断。解决增加响应指纹比对——采集已知404页面样本如/nonexistent12345.html计算其MD5后续所有响应与之比对# 预扫描阶段 sample_404 requests.get(fhttps://{target}/nonexistent_{int(time.time())}.html).content self._404_fingerprint hashlib.md5(sample_404).hexdigest() # 扫描时 for path in wordlist: r session.get(fhttps://{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.29X-Powered-By: PHP/7.2.24SSL证书过期——这四个事实单独看是信息组合起来就是一条高危告警“Apache 2.4.29 PHP 7.2.24 组合存在CVE-2019-0211privilege 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 clock6.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: fSeverity: {alert[severity]}\n\n fRemediation: {alert[remediation]}\n\n fAffected targets:\n \n.join(set(r[target] for r in alert[matched_results])), issuetype: {name: Bug} } } resp requests.post(url, jsonpayload, authauth) if resp.status_code 201: return resp.json()[key] else: logger.error(fJira 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规则变更导致误报工单关闭后漏洞实际仍在。现在我的扫描报告末尾永远有一行加粗提示“所有工单需经人工复核后关闭”。希望帮到你。本文还有配套的精品资源点击获取 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的综合网络安全扫描工具”不能只靠nmaprequests拼凑完事你手头有个需求写个能跑在内网、不依赖云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/负责任务队列、线程/协程池管理、全局配置加载、信号捕获CtrlC中断、日志分级DEBUG/INFO/WARN/ERROR。这里不用asyncio硬上而用concurrent.futures.ThreadPoolExecutorqueue.Queue组合——因为多数扫描动作如Socket连接、HTTP请求本质是I/O阻塞线程比协程更易调试、更少隐式状态。scanner/按协议垂直切分。port_scanner.pyTCP 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不如requestsurllib3可控依赖选用理由替代方案踩坑记录python-libnmap解析nmap XML输出稳定支持增量解析大扫描不爆内存API返回对象而非字符串python-nmap底层调subprocess.Popen易被杀进程导致XML残缺os.system(nmap -oX ...)XML未闭合标签导致xml.etree.ElementTree解析崩溃无重试机制requestsurllib3.util.retry.Retry可精细控制重试策略如对503状态码重试3次对401不重试Session复用连接池verifyFalse时仍能获取原始证书链httpx默认启用HTTP/2某些老旧Web服务器如Apache 2.2返回400 Bad Request且无明确报错排查耗时3小时pyOpenSSL直接解析PEM证书提取notBefore/notAfter/subjectAltName等字段比ssl标准库更易处理自签名证书异常ssl.get_server_certificate()无法获取OCSP stapling状态且对SNI缺失的服务器会握手失败安装命令必须带约束pip install requests2.28.0,2.30.0 urllib31.26.12,1.27.0 pyOpenSSL23.0.0,24.0.0 python-libnmap0.7.0,0.8.0注意requests 2.30.0移除了urllib3.util.retry.Retry的raise_on_statusFalse参数会导致重试逻辑失效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, errorsignore) 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_outputTrue, textTrue) 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, ferror:{str(e)[:20]}, None except ConnectionRefusedError: sock.close() return False, closed, None except Exception as e: sock.close() return False, ferror:{str(e)[:20]}, None逻辑说明第一次1秒超时是主力判断覆盖90%正常网络第二次3秒超时专治高延迟链路如跨省专线第三次用ping辅助判断若ping不通大概率是主机下线unreachable若ping通但端口连不上基本确定是防火墙拦截filtered。banner只在第一次连接成功时尝试获取避免对慢服务造成额外延迟。参数说明timeout2.0是总耗时上限实际执行中三次探测最大耗时约5秒131远低于传统单次5秒扫描banner解码用errorsignore防止二进制数据崩掉生产环境建议用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_workerseffective_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: fexception:{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必须结构化解析并打分Header检查逻辑风险等级业务含义Server正则匹配nginx/[0-9.]、Apache/[0-9.]版本号查CVE数据库高暴露精确中间件版本降低漏洞利用门槛X-Powered-By存在即标记值为PHP/7.4.33等具体版本中同上且常被用于指纹识别Strict-Transport-Security检查max-age是否≥315360001年有includeSubDomains低缺失为中危缺失意味着HTTP→HTTPS降级攻击可行Content-Security-Policy是否存在且default-src不为none或*中CSP缺失或配置过宽XSS风险上升X-Frame-Options值为DENY或SAMEORIGIN低缺失为中危缺失导致点击劫持ClickjackingX-Content-Type-Options值必须为nosniff低缺失为中危缺失可能触发MIME类型嗅探导致JS/CSS被误执行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(rmax-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://..., verifyTrue)只能验证证书是否由可信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), timeout5) as sock: with context.wrap_socket(sock, server_hostnamehostname) as ssock: # 获取原始证书 cert_der ssock.getpeercert(binary_formTrue) 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 bsubjectAltName 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-rate100] else: # 普通用户降级为connect扫描 nmap_args [-sT, -T4]5.2 现象扫描HTTPS站点时requests抛SSLError: certificate verify failed但浏览器能正常访问原因目标使用私有CA证书如企业内部PKI而requests只信任Mozilla CA列表不读取系统证书存储。解决不简单设verifyFalse会丢失证书链信息而是将系统证书路径注入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_ca5.3 现象多线程扫描时urllib3抛Max retries exceeded但单线程正常原因urllib3的PoolManager默认连接池大小为1020个线程共用一个Session时连接复用竞争导致超时。解决为每个线程创建独立Session并显式设置连接池def create_session(): session requests.Session() adapter requests.adapters.HTTPAdapter( pool_connections20, pool_maxsize20, max_retriesurllib3.util.retry.Retry( total3, backoff_factor0.3, status_forcelist[502, 503, 504], ) ) session.mount(http://, adapter) session.mount(https://, adapter) return session5.4 现象扫描结果JSON导出后datetime对象序列化失败原因json.dumps()不支持datetime而扫描时间戳用datetime.now()生成。解决统一用time.time()float替代或自定义JSONEncoderclass ScanJSONEncoder(json.JSONEncoder): def default(self, obj): if isinstance(obj, datetime): return obj.isoformat() return super().default(obj) json.dumps(results, clsScanJSONEncoder)5.5 现象dir_scanner爆破目录时403 Forbidden和404 Not Found返回相同HTML内容无法区分原因某些WAF如Cloudflare对403/404返回相同错误页仅靠状态码和内容长度无法判断。解决增加响应指纹比对——采集已知404页面样本如/nonexistent12345.html计算其MD5后续所有响应与之比对# 预扫描阶段 sample_404 requests.get(fhttps://{target}/nonexistent_{int(time.time())}.html).content self._404_fingerprint hashlib.md5(sample_404).hexdigest() # 扫描时 for path in wordlist: r session.get(fhttps://{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.29X-Powered-By: PHP/7.2.24SSL证书过期——这四个事实单独看是信息组合起来就是一条高危告警“Apache 2.4.29 PHP 7.2.24 组合存在CVE-2019-0211privilege 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 clock6.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: fSeverity: {alert[severity]}\n\n fRemediation: {alert[remediation]}\n\n fAffected targets:\n \n.join(set(r[target] for r in alert[matched_results])), issuetype: {name: Bug} } } resp requests.post(url, jsonpayload, authauth) if resp.status_code 201: return resp.json()[key] else: logger.error(fJira 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规则变更导致误报工单关闭后漏洞实际仍在。现在我的扫描报告末尾永远有一行加粗提示“所有工单需经人工复核后关闭”。希望帮到你。本文还有配套的精品资源点击获取