1. 从“为什么”开始子域名探测的实战价值再思考上次我们聊了用Python实现子域名探测的基础方法主要是基于字典爆破的思路。很多朋友跟着做了一遍反馈说“工具跑起来了但感觉差点意思”。这种感觉是对的。一个能跑起来的脚本和一个在实战中真正好用的工具中间隔着一道巨大的鸿沟。今天这篇我们就来填这道鸿沟聊聊如何把一个基础探测脚本打磨成一个能在真实网络环境中稳定、高效、智能工作的“侦察兵”。为什么需要继续深入因为网络环境不是实验室。你面对的不是一个孤立的靶机而是由CDN、WAF、云防护、负载均衡、以及各种奇葩网络策略构成的复杂战场。一个简单的requests.get可能会因为超时、重定向、证书错误、或者对方服务器的“不友好”策略而失败。更关键的是盲目的爆破不仅效率低下噪音巨大还可能触发安全设备的告警。我们需要的是带着“脑子”去探测。所以本篇的核心不再是“如何发出一个请求”而是“如何设计一套策略让请求更聪明、更隐蔽、更抗干扰”。我们会围绕几个实战中的核心痛点展开如何应对网络超时和异常如何设计智能的重试与并发控制在效率和稳定性间找到平衡如何整合更多的数据源不仅仅是字典来提升发现率以及如何优雅地处理和保存海量的结果数据这些才是一个信息收集工具从“玩具”走向“武器”的关键。2. 构建健壮的请求核心异常处理与会话管理一个脚本在本地测试时完美运行一到实战就各种崩溃十有八九是网络请求层太脆弱。直接使用requests.get()而不做任何包装就像不戴护具上战场。2.1 设计一个抗造的请求函数我们不能让一个子域名的请求失败导致整个程序中断也不能因为一个请求卡住就让所有并发任务等待。我们需要一个自带异常处理和超时控制的请求函数。import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry import socket import time def robust_request(url, timeout5, max_retries3, backoff_factor0.5): 一个健壮的HTTP请求函数具备重试和超时控制。 参数: url: 请求的URL timeout: 连接和读取超时时间秒 max_retries: 最大重试次数针对特定错误 backoff_factor: 重试间隔退避因子 返回: response: 响应对象如果成功 None: 如果所有尝试都失败 # 创建一个自定义会话并配置重试策略 session requests.Session() # 定义重试策略针对连接错误、超时、特定的HTTP状态码如502503504 retry_strategy Retry( totalmax_retries, backoff_factorbackoff_factor, # 重试等待时间 backoff_factor * (2^(重试次数-1)) status_forcelist[500, 502, 503, 504], # 遇到这些状态码会重试 allowed_methods[HEAD, GET, OPTIONS] # 只对这些方法重试 ) # 将重试策略适配到HTTP和HTTPS请求 adapter HTTPAdapter(max_retriesretry_strategy) session.mount(http://, adapter) session.mount(https://, adapter) # 设置全局超时连接读取 session.request lambda method, url, **kwargs: requests.Session.request( session, method, url, timeouttimeout, **kwargs ) try: # 禁用SSL证书验证警告针对自签名证书实战中可根据需要调整 # 注意在高度安全的环境下应妥善处理证书验证 response session.get(url, verifyFalse, allow_redirectsTrue) # 即使请求成功也可能返回非200状态码这里我们一并返回response对象 # 由调用者根据状态码判断是否“发现”了子域名 return response except requests.exceptions.ConnectTimeout: print(f[超时] 连接超时: {url}) except requests.exceptions.ReadTimeout: print(f[超时] 读取超时: {url}) except requests.exceptions.ConnectionError as e: # 这里可能包含多种错误如DNS解析失败、拒绝连接等 # 我们可以根据异常信息进一步分类 if Name or service not known in str(e) or nodename nor servname provided in str(e): # DNS解析失败通常意味着子域名不存在 pass # 安静地跳过不打印错误避免刷屏 elif Connection refused in str(e): print(f[拒绝] 连接被拒绝: {url}) # 端口可能开放但服务拒绝连接 else: print(f[连接错误] {url} - {e}) except requests.exceptions.SSLError: print(f[SSL错误] 证书问题: {url}) # 可能存在自签名证书或SSL配置错误 except socket.gaierror: # socket层的地址信息错误通常也是DNS解析问题 pass except Exception as e: # 捕获其他所有未预见的异常避免程序崩溃 print(f[未知错误] 请求 {url} 时发生错误: {type(e).__name__} - {e}) return None这个robust_request函数做了几件关键事第一它使用了requests.Session配合HTTPAdapter和Retry为请求自动添加了重试机制。这意味着如果因为网络瞬时波动或服务器临时过载返回了503错误它会自动等待片刻后重试而不是直接放弃。第二它用try...except块包裹了所有可能发生的异常从超时、连接错误到SSL问题并对最常见的DNS解析失败子域名不存在进行了静默处理避免错误日志刷屏影响查看有效结果。第三它统一了超时时间并允许灵活配置。在实战中针对不同网络环境如内网、外网你可能需要调整timeout和max_retries参数。注意verifyFalse禁用了SSL证书验证这在实际渗透测试或扫描不信任的目标时常用以避免因证书问题阻断扫描。但在其他合规场景下请谨慎使用或配置正确的证书路径。2.2 会话Session的妙用与连接池你可能会问为什么不用简单的requests.get而要创建一个Session对象这不仅仅是为了重试。Session对象会保持一个连接池。当你向同一主机比如target.com发送大量请求探测其成千上万个子域名时使用Session可以复用底层的TCP连接避免为每个请求都进行“三次握手”和“四次挥手”这能显著降低系统开销和延迟提升扫描速度。特别是在HTTPS场景下建立TLS连接的成本更高复用连接的优势更加明显。3. 效率与资源的博弈并发控制策略有了健壮的请求函数接下来要解决速度问题。单线程顺序请求几万个子域名是不可接受的。我们必须使用并发。但并发不是简单的“开很多线程”它是一场在速度、资源消耗和目标服务器压力之间的精细博弈。3.1 为什么选择线程池而非多进程Python的并发主要有线程threading和进程multiprocessing。对于I/O密集型任务如网络请求大部分时间在等待响应使用多线程是更合适的选择因为线程切换开销远小于进程。虽然Python有GIL全局解释器锁但在I/O操作期间GIL会被释放因此多线程在I/O密集型场景下依然能有效提升性能。我们使用concurrent.futures模块中的ThreadPoolExecutor它提供了高级的线程池接口。3.2 设计可调节的并发控制器直接开1000个线程猛冲很可能导致本地网络拥堵、内存耗尽或者把目标服务器打挂触发防护。我们需要一个可调节的“阀门”。from concurrent.futures import ThreadPoolExecutor, as_completed import queue import threading class SubdomainScanner: def __init__(self, base_domain, wordlist, max_workers50, request_timeout5): self.base_domain base_domain self.wordlist wordlist # 假设这是一个子域名列表 self.max_workers max_workers # 并发线程数 self.request_timeout request_timeout self.found_domains [] self.lock threading.Lock() # 用于线程安全地写入结果列表 self.task_queue queue.Queue() # 任务队列可选另一种管理方式 def _scan_one(self, subdomain): 扫描单个子域名的具体任务 url fhttp://{subdomain}.{self.base_domain} # 使用我们上面定义的健壮请求函数 resp robust_request(url, timeoutself.request_timeout) if resp is not None: # 这里可以根据状态码、响应内容等进一步判断 # 例如有些页面会返回404但域名是存在的有些可能跳转到其他地址 with self.lock: self.found_domains.append({ subdomain: subdomain, url: url, status_code: resp.status_code, final_url: resp.url # 注意重定向后的最终URL }) print(f[] Found: {subdomain}.{self.base_domain} ({resp.status_code}) - {resp.url}) # 如果resp为Nonerobust_request内部已处理日志这里无需操作 def run(self): 启动扫描 print(f[*] 开始扫描 {self.base_domain}, 使用 {self.max_workers} 个并发线程) # 使用ThreadPoolExecutor管理线程池 with ThreadPoolExecutor(max_workersself.max_workers) as executor: # 提交所有任务到线程池 # 这里使用字典将Future对象与子域名关联便于后续处理 future_to_sub { executor.submit(self._scan_one, sub): sub for sub in self.wordlist } # 使用as_completed获取已完成的任务可以实时看到进度 for future in as_completed(future_to_sub): subdomain future_to_sub[future] try: # 这里其实不需要get结果因为结果已存储在self.found_domains中 # 调用future.result()是为了触发可能未被捕获的异常 future.result() except Exception as exc: # 理论上_scan_one里的异常已被robust_request处理这里捕获的是其他意外 print(f{subdomain} 在扫描中产生了异常: {exc}) print(f[*] 扫描结束。共发现 {len(self.found_domains)} 个有效子域名。) return self.found_domains这个扫描器类SubdomainScanner的核心是run方法。它创建了一个固定大小的线程池max_workers然后将每个子域名的探测任务_scan_one提交给线程池。ThreadPoolExecutor会自动管理线程的创建、调度和回收。as_completed迭代器会按照任务完成的顺序返回结果让我们能够实时看到进度而不是等到所有任务结束。关键参数max_workers的设置经验这个值不是越大越好。设置过大如500会导致大量线程争抢CPU和网络资源上下文切换开销剧增可能反而变慢甚至导致程序不稳定。一般建议设置在50-200之间具体取决于你的本地网络带宽、CPU核心数和目标服务器的承受能力。对于公开的互联网目标建议从较低值如30开始测试观察本地资源占用和目标响应情况再逐步调整。4. 超越字典爆破多源情报融合策略纯字典爆破的瓶颈在于字典的质量和广度。一个精心维护的字典可能包含几十万条记录但互联网是动态的每天都有新的子域名被创建。我们需要引入外部情报实现“字典爆破 智能枚举”的组合拳。4.1 集成证书透明度CT日志查询证书透明度Certificate Transparency, CT是一个公开的日志系统要求CA公开记录颁发的每一张SSL/TLS证书。这些日志里包含了证书对应的域名信息是发现子域名的金矿。我们可以通过查询CT日志的公开API来获取目标域名的所有已证书域名。import requests import json def query_cert_transparency(domain): 通过 crt.sh 查询证书透明度日志发现子域名。 crt.sh 是一个聚合了多个CT日志的搜索引擎。 found set() # 使用集合自动去重 api_url fhttps://crt.sh/json?q%.{domain}excludeexpired try: resp robust_request(api_url, timeout10) if resp and resp.status_code 200: data resp.json() for item in data: # 提取 common_name 和 name_value 字段 common_name item.get(common_name, ) name_value item.get(name_value, ) # 处理 name_value它可能是一个字符串也可能是一个包含多个域名的换行分隔字符串 names [] if common_name: names.append(common_name) if name_value: # 如果name_value是字符串且包含换行则拆分 if isinstance(name_value, str): names.extend([n.strip() for n in name_value.split(\n) if n.strip()]) elif isinstance(name_value, list): names.extend(name_value) # 过滤出属于目标域名的子域名 for name in names: name name.lower().strip() # 处理通配符证书如 *.dev.target.com if name.startswith(*.): name name[2:] # 检查是否以目标域名结尾 if name.endswith(f.{domain}) or name domain: found.add(name) else: print(f[!] 查询 crt.sh API 失败: {resp.status_code if resp else 无响应}) except json.JSONDecodeError: print(f[!] 解析 crt.sh 返回的JSON数据失败) except Exception as e: print(f[!] 查询证书透明度日志时发生未知错误: {e}) # 将结果转换为标准的子域名前缀格式去除主域名部分 subdomains [] for full_domain in found: if full_domain domain: subdomains.append() # 用表示根域名 elif full_domain.endswith(f.{domain}): sub_prefix full_domain[:-len(domain)-1] # 去掉末尾的 .domain subdomains.append(sub_prefix) return subdomains这个函数通过crt.sh的公开API进行查询。它返回的数据中可能包含通配符证书*.dev.target.com和多个域名SAN主题备用名称。我们需要仔细解析这些数据提取出所有属于目标域名的子域名。CT日志的优点是覆盖面广能发现很多非常规的、不在常见字典里的子域名例如内部系统临时对外暴露时申请的证书。实战心得CT查询的结果可能会有重复和大量“脏数据”如过期的、测试的域名需要结合后续的存活验证进行过滤。同时API可能有速率限制在批量扫描时需要注意添加延时。4.2 集成DNS查询AXFR/区域传输漏洞检测如果目标域名配置不当其DNS服务器可能允许“区域传输”AXFR请求。这将允许任何人获取该域的所有DNS记录包括所有子域名。这是一种经典的配置错误漏洞检测。import dns.resolver import dns.zone import dns.exception def try_dns_axfr(domain, nameserverNone): 尝试对目标域名进行DNS区域传输。 参数: domain: 目标域名 nameserver: 指定的DNS服务器如果为None则使用系统默认 found_subdomains [] # 如果没有指定nameserver先尝试获取目标的权威DNS服务器 ns_servers [] if not nameserver: try: answers dns.resolver.resolve(domain, NS) ns_servers [str(rdata) for rdata in answers] print(f[*] 发现 {domain} 的权威DNS服务器: {ns_servers}) except (dns.resolver.NoAnswer, dns.resolver.NXDOMAIN, dns.exception.Timeout): print(f[!] 无法解析 {domain} 的NS记录将使用系统默认解析器。) ns_servers [None] # 使用默认 else: ns_servers [nameserver] for ns in ns_servers: print(f[*] 尝试向 {ns if ns else 系统默认} 请求 {domain} 的区域传输...) try: if ns: # 创建指定NS的解析器 resolver dns.resolver.Resolver() resolver.nameservers [ns] # 注意dns.query.xfr 需要IP地址所以需要先将NS主机名解析为IP try: ns_ip dns.resolver.resolve(ns, A)[0].address except: print(f[!] 无法解析NS服务器 {ns} 的IP地址跳过。) continue else: # 使用系统默认这里简化处理实际区域传输通常需要指定权威服务器 print(f[!] 未指定权威NS区域传输成功率极低跳过。) continue # 执行区域传输 # 注意dns.zone.from_xfr 需要网络操作这里使用一个简化的替代方案尝试查询常见的记录类型 # 更严谨的做法需要使用像dnspython的xfr方法或专用工具如dig # 此处为演示逻辑实际应用中建议集成命令行工具 dig AXFR {ns_ip} {domain} 的结果解析 print(f[!] 提示完整的AXFR检测通常需调用系统dig命令或使用更底层的socket实现。) print(f[!] 可执行: dig AXFR {ns_ip} {domain}) # 以下代码块仅为逻辑示意实际不可直接运行 # zone dns.zone.from_xfr(dns.query.xfr(ns_ip, domain)) # for name, node in zone.nodes.items(): # found_subdomains.append(str(name)) except dns.exception.FormError: print(f[-] 服务器 {ns} 拒绝区域传输请求正常情况。) except dns.exception.Timeout: print(f[-] 连接DNS服务器 {ns} 超时。) except Exception as e: print(f[!] 区域传输尝试时发生错误 ({ns}): {type(e).__name__}) return found_subdomains # 实际应用中这里应返回从AXFR获取的子域名列表由于Python的dnspython库进行完整的AXFR请求相对复杂上述代码更多是展示检测逻辑和流程。在实际工具中我通常选择封装系统调用例如使用subprocess模块调用dig或nslookup命令来执行AXFR请求并解析输出。关键点区域传输漏洞现在已不常见但一旦发现就是“宝藏”因为它能直接列出所有子域名记录包括那些非常隐蔽的内部域名。测试时务必先获得授权。4.3 融合多源数据与去重当我们从字典、CT日志、甚至其他来源如搜索引擎爬取、第三方子域名数据库API获取到子域名列表后第一步是合并和去重。def aggregate_subdomains(base_domain, dict_subs, ct_subs, other_subs[]): 聚合来自不同来源的子域名并进行初步清洗。 返回: 一个去重后的子域名前缀列表。 all_subs_set set() # 添加字典来源 for sub in dict_subs: all_subs_set.add(sub.strip().lower()) # 添加CT日志来源 for sub in ct_subs: all_subs_set.add(sub.strip().lower()) # 添加其他来源 for sub in other_subs: all_subs_set.add(sub.strip().lower()) # 清洗数据移除空字符串、移除可能包含协议头或路径的脏数据 cleaned_subs [] for sub in all_subs_set: if not sub: continue # 如果sub是完整域名提取子域名部分 if sub.endswith(f.{base_domain}): sub_prefix sub[:-len(base_domain)-1] cleaned_subs.append(sub_prefix) elif sub base_domain: cleaned_subs.append() # 根域名 elif . not in sub: # 纯子域名前缀 cleaned_subs.append(sub) # 其他格式的脏数据暂时忽略 else: # 可能是不相关的域名例如CT日志里混入了其他域名 pass # 最终去重并返回 return list(set(cleaned_subs))聚合后我们得到一个庞大的待检测列表。接下来在并发探测之前还可以做一个优化DNS预解析过滤。对于海量子域名直接进行HTTP/HTTPS请求成本很高。我们可以先批量进行DNS A记录解析快速过滤掉大量不存在的域名只对能解析出IP的域名进行后续的Web请求。这能极大减少无效的网络流量和扫描时间。5. 结果处理与报告生成让数据说话探测出子域名只是第一步如何组织、分析和呈现这些结果决定了信息的价值。我们不能只输出一个简单的文本列表。5.1 结构化存储与去重在扫描过程中我们收集了子域名、状态码、最终URL等信息。应该将这些信息实时保存到结构化的文件中比如JSON或CSV避免程序意外中断导致数据丢失。import json import csv from datetime import datetime class ResultManager: def __init__(self, base_domain): self.base_domain base_domain self.results [] # 存储每次扫描的结果字典 self.timestamp datetime.now().strftime(%Y%m%d_%H%M%S) self.filename_json fsubdomains_{base_domain}_{self.timestamp}.json self.filename_csv fsubdomains_{base_domain}_{self.timestamp}.csv def add(self, subdomain_info): 添加一条扫描结果 self.results.append(subdomain_info) # 可以选择实时写入文件或扫描结束后统一写入 # 实时写入更安全但频繁IO可能影响性能。这里采用批量写入。 def save(self): 将结果保存到JSON和CSV文件 if not self.results: print([!] 无结果可保存。) return # 保存为JSON便于程序后续读取 try: with open(self.filename_json, w, encodingutf-8) as f: json.dump({ base_domain: self.base_domain, scan_time: self.timestamp, count: len(self.results), results: self.results }, f, indent2, ensure_asciiFalse) print(f[*] 结果已保存至 JSON 文件: {self.filename_json}) except IOError as e: print(f[!] 保存JSON文件失败: {e}) # 保存为CSV便于用Excel打开查看和分析 try: # 定义CSV表头 fieldnames [subdomain, full_domain, status_code, final_url, title, ip_address] with open(self.filename_csv, w, newline, encodingutf-8) as f: writer csv.DictWriter(f, fieldnamesfieldnames) writer.writeheader() for res in self.results: # 确保结果字典包含所有字段没有的填空 row {field: res.get(field, ) for field in fieldnames} writer.writerow(row) print(f[*] 结果已保存至 CSV 文件: {self.filename_csv}) except IOError as e: print(f[!] 保存CSV文件失败: {e}) def load(self, filename): 从JSON文件加载历史结果 try: with open(filename, r, encodingutf-8) as f: data json.load(f) self.results data.get(results, []) print(f[*] 从 {filename} 加载了 {len(self.results)} 条历史结果。) return self.results except Exception as e: print(f[!] 加载结果文件失败: {e}) return []ResultManager类负责结果的收集、存储和加载。使用时间戳命名文件可以避免覆盖之前的扫描结果。JSON格式保留了完整的结构化信息方便后续进行深度分析或导入其他工具。CSV格式则提供了极佳的可读性和表格处理能力。实战技巧在扫描过程中除了状态码和最终URL强烈建议尝试提取页面的title标签内容。一个独特的标题如“OA登录”、“数据库管理”、“测试环境”能让你快速识别出有价值的目标。这可以通过在robust_request成功后用BeautifulSoup或正则表达式简单解析响应HTML来实现。5.2 基础分析与报告有了结构化的数据我们可以做一些简单的自动分析生成更有洞察力的报告。def generate_summary_report(results): 根据扫描结果生成简易的文本报告。 if not results: return 无扫描结果。 total len(results) status_2xx sum(1 for r in results if 200 r.get(status_code, 0) 300) status_3xx sum(1 for r in results if 300 r.get(status_code, 0) 400) status_4xx sum(1 for r in results if 400 r.get(status_code, 0) 500) status_5xx sum(1 for r in results if 500 r.get(status_code, 0) 600) status_other total - (status_2xx status_3xx status_4xx status_5xx) # 按状态码分类 report_lines [] report_lines.append(*50) report_lines.append(子域名扫描结果摘要) report_lines.append(*50) report_lines.append(f总计发现: {total}) report_lines.append(f状态码分布:) report_lines.append(f - 2xx (成功): {status_2xx}) report_lines.append(f - 3xx (重定向): {status_3xx}) report_lines.append(f - 4xx (客户端错误): {status_4xx}) report_lines.append(f - 5xx (服务器错误): {status_5xx}) report_lines.append(f - 其他/未知: {status_other}) report_lines.append() # 列出所有成功的2xx子域名通常是最值得关注的 if status_2xx 0: report_lines.append([] 可访问的子域名 (2xx):) for r in results: if 200 r.get(status_code, 0) 300: report_lines.append(f - {r.get(full_domain)} ({r.get(status_code)}) - {r.get(final_url)}) report_lines.append() # 列出重定向目标可能暴露内部架构 if status_3xx 0: report_lines.append([] 重定向子域名 (3xx):) redirect_map {} for r in results: if 300 r.get(status_code, 0) 400: orig r.get(full_domain) final r.get(final_url, ) if final and final ! fhttp://{orig} and final ! fhttps://{orig}: if final not in redirect_map: redirect_map[final] [] redirect_map[final].append(orig) for final_url, origins in redirect_map.items(): report_lines.append(f - {final_url}) for orig in origins[:3]: # 只显示前3个来源避免太长 report_lines.append(f 来自: {orig}) if len(origins) 3: report_lines.append(f ... 以及另外 {len(origins)-3} 个) report_lines.append() return \n.join(report_lines)这个报告函数提供了状态码分布概览并重点突出了两类关键目标一是返回2xx成功状态码的子域名它们很可能正在运行Web服务二是发生了重定向的子域名特别是多个子域名重定向到同一个最终地址的情况这可能暗示着负载均衡策略或统一的登录入口。进阶方向你可以进一步集成IP地址解析、端口扫描针对解析出的IP、WAF指纹识别、或与漏洞扫描器联动将子域名探测作为更广泛攻击面管理流程的起点。6. 实战中的“坑”与优化技巧把上面的模块组装起来你已经有了一个相当强大的子域名扫描器。但在真实环境中运行还会遇到一些意想不到的问题。这里分享几个我踩过的坑和对应的优化技巧。坑一DNS解析缓存与超时。Python的socket库或requests底层使用的DNS解析可能有缓存或者解析速度慢。当并发量很高时DNS查询可能成为瓶颈甚至超时。解决方案使用dnspython库并配置自己的DNS解析器可以设置更短的超时时间和自定义的DNS服务器如8.8.8.8,1.1.1.1。对于批量探测可以先进行一次大规模的DNS A记录解析将域名-IP的映射关系缓存下来后续的HTTP请求直接使用IP地址注意处理HTTPS的SNI问题。坑二目标网站的防护与封禁。高频、规律的请求很容易被WAF或防护设备识别为扫描行为导致你的IP被临时封禁。解决方案第一降低并发数max_workers。第二在请求之间加入随机延时time.sleep(random.uniform(0.5, 2))。第三随机化User-Agent头模拟不同浏览器的请求。第四如果条件允许使用代理IP池来分散请求源。坑三海量结果中的噪音。你会探测到大量返回4xx/5xx状态码、或者内容为默认错误页、域名停放页的子域名。这些信息价值较低。解决方案在保存结果时进行初步过滤。例如可以忽略状态码为404、且页面内容长度小于特定值、或包含特定关键词如“parking”、“未找到”的结果。只保留状态码为2xx/3xx/403/500可能有信息泄露等有价值的条目。这需要在robust_request函数中增加对响应内容的简单分析。坑四内存与性能。如果字典非常大上百万条一次性加载到内存并提交所有任务到线程池可能导致内存消耗过高。解决方案使用生产者-消费者模型。一个线程负责从文件或生成器中读取子域名生产者放入一个队列queue.Queue线程池中的工作线程消费者从队列中获取任务执行。这样可以控制内存中的待处理任务数量。同时对于超大的字典可以考虑先进行DNS预过滤大幅减少需要HTTP探测的任务量。一个简单的生产者-消费者模型示意import queue import threading import time def producer(wordlist_path, task_queue, max_queued1000): 生产者从文件读取子域名放入队列 with open(wordlist_path, r, encodingutf-8, errorsignore) as f: for line in f: sub line.strip() if sub and not sub.startswith(#): # 跳过空行和注释 # 如果队列太满生产者暂停 while task_queue.qsize() max_queued: time.sleep(0.1) task_queue.put(sub) # 放入结束信号 for _ in range(consumer_count): task_queue.put(None) def consumer(base_domain, task_queue, result_manager): 消费者从队列获取子域名并进行探测 while True: sub task_queue.get() if sub is None: # 收到结束信号 task_queue.put(None) # 放回信号让其他消费者也能结束 break # 调用扫描函数 url fhttp://{sub}.{base_domain} resp robust_request(url) if resp: result_manager.add({ subdomain: sub, full_domain: f{sub}.{base_domain}, status_code: resp.status_code, final_url: resp.url }) task_queue.task_done()最后也是最重要的伦理与法律提醒子域名探测如同网络空间的“敲门”必须在获得明确授权的前提下进行。未经授权对他人网络资产进行扫描不仅是非法的也可能对目标系统造成不必要的负载甚至破坏。请务必在合规的环境如自己的测试环境、授权的渗透测试项目、漏洞众测平台中使用这些技术。技术的价值在于建设和保护而非破坏。