1. 项目概述从脚本小子到专业工具的跨越上次我们聊了用Python实现子域名探测的基础方法主要是基于字典的暴力枚举。那篇文章发布后很多朋友留言说“学到了”、“思路清晰”但也有人提出“效率是不是有点低”、“遇到有防护的网站怎么办”。问得好这说明大家已经不满足于“能用”开始追求“好用”和“稳健”了。这正是我们这次要深入探讨的。“子域名探测二”这个标题本身就意味着进阶。如果说第一部分是教你造一把能用的“刀”那么第二部分就是教你如何把这把刀打磨得更锋利、更趁手甚至给它配上“刀鞘”和“磨刀石”。在实战的信息收集阶段子域名枚举的广度和深度直接决定了后续渗透测试的入口点数量。一个被遗漏的子域名可能就是整个安全防线的突破口。因此提升探测的自动化、智能化和健壮性是每个安全从业者或爱好者的必修课。这次我们将不再局限于一个简单的requests循环。我们会引入异步并发来大幅提升速度整合多个数据源如证书透明度日志、搜索引擎语法来扩大探测面并设计完善的错误处理和结果去重机制让脚本从一个“实验品”蜕变为一个可以融入实际工作流的“工具”。无论你是安全工程师、红队成员还是对网络空间测绘感兴趣的研究者这套思路和代码都能为你提供直接的参考价值。我们用的语言依然是Python因为它生态丰富、库强大能让我们快速实现想法。2. 核心思路与架构升级在第一部分中我们的脚本结构可以概括为读取字典 - 拼接域名 - 发起HTTP请求 - 根据状态码判断存活。这个模型简单直接但瓶颈也很明显速度慢同步请求、覆盖面窄仅依赖字典、容错差网络波动可能导致脚本中断。2.1 设计目标与方案选型本次升级我们瞄准三个核心目标速度利用异步I/Oasyncio aiohttp或线程池将网络请求的等待时间重叠起来实现并发探测效率提升十倍甚至百倍。广度不单靠字典爆破引入被动信息源进行交叉验证和补充。我们主要考虑证书透明度CT日志很多网站会为子域名申请SSL证书这些申请记录是公开的宝藏。搜索引擎聚合利用公开的搜索引擎接口或语法搜集已被收录的子域名。DNS记录查询尝试查询常见的DNS记录类型如A, AAAA, CNAME, MX有时能发现非Web服务的子域名。健壮性完善的异常处理、超时控制、重试机制以及最终结果的自动去重与格式化输出。为什么选择这样的架构因为在实战中时间和信息的全面性是关键。一个目标可能只有几十个常见子域名也可能有成千上万个。同步脚本在后者面前几乎不可用。同时仅靠字典会漏掉那些不常见、但确实存在的子域名例如为某个临时活动创建的campaign2023.target.com而CT日志和DNS查询恰好能弥补这一点。2.2 技术栈与工具链我们将主要依赖以下Python库aiohttp与asyncio用于实现高性能的异步HTTP/HTTPS请求。这是提升速度的核心。aiodns用于异步DNS查询可以快速解析大量域名。requests虽然我们用aiohttp做并发请求但在一些简单的、非并发的API调用如查询某些CT日志接口时它依然稳定可靠。tqdm为我们的循环加上一个美观的进度条实时了解任务进度体验更友好。内置库argparse处理命令行参数、json解析API返回数据、logging记录运行日志便于排查问题。注意异步编程asyncio有一定的学习曲线但其带来的性能收益是巨大的。如果你对异步还不熟悉可以把本章的代码当作一个很好的学习案例。我们也会尽量解释清楚关键步骤。3. 模块化实现与核心代码解析我们将把功能拆分成独立的模块这样代码更清晰也便于后期维护和扩展。整个项目结构可以规划如下subdomain_scanner_v2/ ├── scanner.py # 主程序入口协调各个模块 ├── modules/ │ ├── __init__.py │ ├── bruteforce.py # 基于字典的异步暴力枚举 │ ├── ct_logs.py # 查询证书透明度日志 │ ├── search_engines.py # 从搜索引擎聚合数据 │ └── dns_query.py # 异步DNS记录查询 ├── utils/ │ ├── __init__.py │ ├── helpers.py # 通用工具函数去重、保存等 │ └── logger.py # 日志配置 └── wordlists/ # 存放各种子域名字典文件 └── common_subdomains.txt3.1 异步暴力枚举模块 (modules/bruteforce.py)这是传统方法的异步升级版。核心思想是创建一个异步的HTTP客户端会话然后用这个会话并发地去请求所有可能的子域名。import aiohttp import asyncio from tqdm.asyncio import tqdm_asyncio from urllib.parse import urljoin async def check_subdomain(session, url, semaphore): 检查单个子域名是否存活返回可访问的URL。 使用信号量(semaphore)控制并发度避免对目标服务器造成过大压力。 async with semaphore: # 控制并发数 try: # 设置超时避免单个请求卡住整个程序 timeout aiohttp.ClientTimeout(total10) async with session.get(url, timeouttimeout, sslFalse, allow_redirectsTrue) as response: # 通常2xx和3xx状态码表示存活401/403等也可能表示服务存在 if response.status 400: # 返回最终跳转后的URL更有价值 final_url str(response.url) return final_url except aiohttp.ClientConnectorError: pass # 连接错误子域名可能不存在或无法访问 except aiohttp.ClientResponseError: pass # 响应错误 except asyncio.TimeoutError: pass # 请求超时 except Exception as e: # 其他异常记录日志但不中断程序 # logging.debug(fError checking {url}: {e}) pass return None async def async_bruteforce(domain, wordlist_path, max_concurrency100): 异步暴力枚举子域名。 :param domain: 主域名如 example.com :param wordlist_path: 字典文件路径 :param max_concurrency: 最大并发数 :return: 存活子域名URL列表 alive_urls [] # 读取字典文件 try: with open(wordlist_path, r, encodingutf-8, errorsignore) as f: subdomains [line.strip() for line in f if line.strip()] except FileNotFoundError: print(f[!] 字典文件未找到: {wordlist_path}) return alive_urls print(f[*] 加载字典完毕共 {len(subdomains)} 条记录开始异步枚举...) # 创建TCP连接器优化连接池 connector aiohttp.TCPConnector(limitmax_concurrency, sslFalse) # 创建信号量控制并发数 semaphore asyncio.Semaphore(max_concurrency) async with aiohttp.ClientSession(connectorconnector) as session: tasks [] for sub in subdomains: # 构造完整的URL尝试HTTP和HTTPS full_domain f{sub}.{domain} for scheme in [http://, https://]: url scheme full_domain # 为每个URL创建一个检查任务 task asyncio.create_task(check_subdomain(session, url, semaphore)) tasks.append(task) # 使用tqdm包装任务显示进度 for task in tqdm_asyncio.as_completed(tasks, totallen(tasks), desc枚举进度): result await task if result: alive_urls.append(result) return alive_urls关键点解析信号量 (asyncio.Semaphore)这是控制“同时进行”的请求数量的关键。如果不加控制瞬间发起成千上万个连接可能会被目标封IP也可能耗尽本地资源。max_concurrency建议设置在50-200之间根据网络情况和目标承受能力调整。连接器 (aiohttp.TCPConnector)它管理底层的TCP连接池复用连接可以避免频繁的三次握手进一步提升速度。limit参数和信号量数值保持一致即可。异常处理我们捕获了多种异常。ClientConnectorError通常代表网络不通或域名不解析TimeoutError代表请求超时。将这些情况静默处理pass只收集成功的响应保证了程序的流畅运行。HTTPS与重定向我们同时尝试了HTTP和HTTPS。设置allow_redirectsTrue让客户端自动跟随跳转并记录最终的URLresponse.url这对于识别CDN、负载均衡器或隐藏的真实路径非常有用。3.2 证书透明度日志查询模块 (modules/ct_logs.py)证书透明度Certificate Transparency是一个公开的日志系统旨在监测和审计SSL证书的签发。我们可以从公开的CT日志接口中查询为某个主域名及其子域名签发的所有证书。import requests import json from urllib.parse import urlparse def query_ct_logs(domain): 查询证书透明度日志获取子域名。 使用 crt.sh 的公开API。 :param domain: 主域名 :return: 子域名集合 subdomains set() url fhttps://crt.sh/json?q%.{domain}excludeexpired try: # 注意这里使用同步requests因为只调用一次API且API返回可能较慢。 # 在主程序中可以将此函数放入线程池以不阻塞主循环。 resp requests.get(url, timeout30) if resp.status_code 200: data resp.json() for entry in data: # 提取证书中的常见名称(CN)和主题备用名称(SAN) common_name entry.get(common_name, ) name_value entry.get(name_value, ) # 处理 common_name if common_name and domain in common_name: # 清理通配符证书 if common_name.startswith(*.): subdomains.add(common_name[2:]) else: subdomains.add(common_name) # 处理 name_value (SAN字段可能包含多个域名用换行分隔) if name_value: for name in name_value.split(\n): name name.strip() if domain in name: if name.startswith(*.): subdomains.add(name[2:]) else: subdomains.add(name) else: print(f[!] crt.sh API 请求失败状态码: {resp.status_code}) except requests.exceptions.RequestException as e: print(f[!] 查询CT日志时发生网络错误: {e}) except json.JSONDecodeError: print(f[!] 解析crt.sh返回的JSON数据失败) # 过滤掉非目标域名的结果因为API是模糊查询 filtered_subs {sub for sub in subdomains if sub.endswith(. domain)} return filtered_subs实操心得数据源选择crt.sh是一个聚合了多个CT日志的免费网站其API简单易用。除此之外还可以考虑Facebook的CT查询库、Google的CT API等但crt.sh对于个人和小规模使用通常足够了。数据清洗API返回的数据可能包含通配符证书*.example.com、主域名本身以及其他不相关的域名。代码中通过检查域名结尾sub.endswith(. domain)进行了过滤这是关键一步。性能考虑这个函数是同步的、网络I/O密集型的。如果在主程序中同步调用可能会阻塞。一个优化方案是使用concurrent.futures.ThreadPoolExecutor将其放入线程池中运行这样它就不会阻塞异步主循环。3.3 搜索引擎聚合模块 (modules/search_engines.py)我们可以利用公开的搜索引擎语法来搜索已被收录的、属于目标域名的页面。这里以Google Custom Search JSON API和SecurityTrails部分功能免费为例。请注意直接爬取搜索引擎页面违反其服务条款且容易被封使用官方API是正确途径。import requests import time def query_google_cse(domain, api_key, cse_id, max_results100): 使用Google自定义搜索API查询子域名。 需要先在Google Cloud Platform创建自定义搜索引擎并获取API Key和CSE ID。 subdomains set() base_url https://www.googleapis.com/customsearch/v1 query fsite:*.{domain} # 搜索语法site:*.example.com start_index 1 while start_index max_results: params { key: api_key, cx: cse_id, q: query, start: start_index, num: 10 # 每页最多10条 } try: resp requests.get(base_url, paramsparams, timeout15) data resp.json() if items in data: for item in data[items]: # 从返回的链接中提取子域名 link item.get(link, ) if link: # 简单提取实际可能需要更复杂的解析 netloc urlparse(link).netloc if domain in netloc: subdomains.add(netloc) else: # 可能没有更多结果了或者触发了速率限制 if error in data: print(f[!] Google API 错误: {data[error].get(message)}) break # 检查是否还有下一页 if queries in data and nextPage in data[queries]: start_index 10 time.sleep(1) # 礼貌性延迟避免请求过快 else: break except Exception as e: print(f[!] 查询Google CSE时出错: {e}) break return subdomains # 另一个例子使用SecurityTrails的API需注册获取API Key def query_securitytrails(domain, api_key): 查询SecurityTrails的子域名数据需要API Key headers {APIKEY: api_key} url fhttps://api.securitytrails.com/v1/domain/{domain}/subdomains try: resp requests.get(url, headersheaders, timeout15) if resp.status_code 200: data resp.json() # SecurityTrails返回的是子域名前缀列表 subdomains {f{sub}.{domain} for sub in data.get(subdomains, [])} return subdomains else: print(f[!] SecurityTrails API 错误: {resp.status_code}) except Exception as e: print(f[!] 查询SecurityTrails时出错: {e}) return set()重要提示使用搜索引擎API通常有每日免费配额限制并且需要注册和配置如Google CSE需要创建项目、启用API、创建自定义搜索引擎。请务必遵守相关API的使用条款。在脚本中建议将API Key等敏感信息通过环境变量或配置文件传入切勿硬编码在代码中。3.4 异步DNS查询模块 (modules/dns_query.py)有些子域名可能没有Web服务HTTP/HTTPS但存在DNS记录如邮件服务器MX记录、别名CNAME记录等。通过DNS查询可以发现它们。import asyncio import aiodns from tqdm.asyncio import tqdm_asyncio async def query_dns_record(resolver, subdomain, record_typeA): 异步查询指定子域名的特定DNS记录。 try: # aiodns 支持 A, AAAA, CNAME, MX, TXT, NS, SOA 等记录类型 result await resolver.query(subdomain, record_type) # 返回查询到的记录列表 return record_type, subdomain, [r.host if hasattr(r, host) else str(r) for r in result] except aiodns.error.DNSError as e: # 查询失败如NXDOMAIN返回None return record_type, subdomain, None except Exception as e: # 其他异常 # logging.debug(fDNS query error for {subdomain} ({record_type}): {e}) return record_type, subdomain, None async def async_dns_enumeration(domain, wordlist_path, record_types[A, AAAA, CNAME, MX]): 异步DNS枚举。 :param domain: 主域名 :param wordlist_path: 字典文件路径 :param record_types: 要查询的DNS记录类型列表 :return: 字典键为子域名值为另一个字典{记录类型: 记录值列表} results {} # 读取字典 try: with open(wordlist_path, r) as f: subdomain_prefixes [line.strip() for line in f if line.strip()] except FileNotFoundError: return results # 创建DNS解析器 resolver aiodns.DNSResolver() tasks [] for prefix in subdomain_prefixes: sub f{prefix}.{domain} for rtype in record_types: task asyncio.create_task(query_dns_record(resolver, sub, rtype)) tasks.append(task) # 处理结果 for task in tqdm_asyncio.as_completed(tasks, totallen(tasks), descDNS查询): rtype, subdomain, records await task if records: # 如果有记录说明子域名存在 if subdomain not in results: results[subdomain] {} results[subdomain][rtype] records return results注意事项DNS查询速度极快相比于HTTP请求DNS查询的延迟低得多因此即使并发数很高对服务器压力也较小但也要注意不要对公共DNS服务器如8.8.8.8进行滥用。记录类型选择A和AAAA记录对应IPv4和IPv6地址是最直接的。CNAME记录别名可能指向另一个域名这有时能揭示内部架构或第三方服务。MX记录指向邮件服务器是重要的资产发现点。结果解析aiodns返回的对象需要根据记录类型提取信息如A记录的host属性是IP地址。代码中做了简单处理实际应用可能需要更精细的解析。4. 主程序调度与结果整合 (scanner.py)现在我们需要一个“大脑”来协调上述所有模块处理命令行参数管理任务执行顺序并整合、去重、输出最终结果。import asyncio import argparse from modules import bruteforce, ct_logs, search_engines, dns_query from utils.helpers import save_results, remove_duplicates import logging # 配置日志 logging.basicConfig(levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s) logger logging.getLogger(__name__) async def main(): parser argparse.ArgumentParser(description高级子域名探测工具 v2.0) parser.add_argument(-d, --domain, requiredTrue, help目标主域名 (例如: example.com)) parser.add_argument(-w, --wordlist, defaultwordlists/common_subdomains.txt, help子域名字典文件路径) parser.add_argument(-o, --output, defaultresults.txt, help结果输出文件) parser.add_argument(-c, --concurrency, typeint, default100, helpHTTP请求并发数) parser.add_argument(--no-bruteforce, actionstore_true, help禁用暴力枚举) parser.add_argument(--no-ct, actionstore_true, help禁用证书透明度查询) parser.add_argument(--no-dns, actionstore_true, help禁用DNS枚举) # 搜索引擎API参数 parser.add_argument(--google-api-key, helpGoogle Custom Search API Key) parser.add_argument(--google-cse-id, helpGoogle Custom Search Engine ID) parser.add_argument(--securitytrails-key, helpSecurityTrails API Key) args parser.parse_args() all_subdomains set() # 使用集合自动去重 logger.info(f开始对目标 {args.domain} 进行子域名探测) # 1. 证书透明度查询 (同步可放入线程池) if not args.no_ct: logger.info([*] 正在查询证书透明度(CT)日志...) ct_subs ct_logs.query_ct_logs(args.domain) all_subdomains.update(ct_subs) logger.info(f[] 从CT日志中发现 {len(ct_subs)} 个子域名) # 2. 搜索引擎聚合 (同步可放入线程池) if args.google_api_key and args.google_cse_id: logger.info([*] 正在通过Google自定义搜索查询...) google_subs search_engines.query_google_cse(args.domain, args.google_api_key, args.google_cse_id) all_subdomains.update(google_subs) logger.info(f[] 从Google搜索中发现 {len(google_subs)} 个子域名) if args.securitytrails_key: logger.info([*] 正在通过SecurityTrails查询...) st_subs search_engines.query_securitytrails(args.domain, args.securitytrails_key) all_subdomains.update(st_subs) logger.info(f[] 从SecurityTrails中发现 {len(st_subs)} 个子域名) # 将收集到的子域名作为基础用于后续的HTTP和DNS检查 # 同时我们仍然会使用字典进行暴力枚举 # 这里可以做一个合并将已发现的子域名前缀加入到一个增强字典中用于DNS枚举 enhanced_prefixes set() for sub in all_subdomains: prefix sub.replace(f.{args.domain}, ) enhanced_prefixes.add(prefix) # 3. 异步DNS枚举 if not args.no_dns: logger.info([*] 开始异步DNS枚举...) dns_results await dns_query.async_dns_enumeration(args.domain, args.wordlist) # dns_results 是字典我们需要提取子域名 dns_subs set(dns_results.keys()) all_subdomains.update(dns_subs) logger.info(f[] 通过DNS枚举发现 {len(dns_subs)} 个子域名) # 可以选择将DNS结果也保存下来 save_dns_details(dns_results, dns_details.json) # 4. 异步HTTP暴力枚举 (对已发现和字典中的子域名进行存活验证) if not args.no_bruteforce: logger.info([*] 开始异步HTTP存活验证...) # 这里我们可以选择对 all_subdomains 中的进行验证也可以继续用字典爆破 # 为了全面我们依然使用原始字典进行爆破 alive_urls await bruteforce.async_bruteforce(args.domain, args.wordlist, args.concurrency) # 从存活的URL中提取子域名 alive_subs set() for url in alive_urls: parsed urlparse(url) alive_subs.add(parsed.netloc) # netloc 是 hostname:port all_subdomains.update(alive_subs) logger.info(f[] 通过HTTP验证发现 {len(alive_subs)} 个存活子域名) save_results(alive_urls, alive_urls.txt) # 最终结果去重和保存 logger.info([*] 整合并去重最终结果...) final_subdomains remove_duplicates(all_subdomains, args.domain) logger.info(f[] 探测完成共发现 {len(final_subdomains)} 个唯一子域名。) save_results(sorted(final_subdomains), args.output) logger.info(f[] 结果已保存至 {args.output}) if __name__ __main__: asyncio.run(main())核心调度逻辑被动收集优先首先进行CT日志和搜索引擎查询。这些是“被动”信息收集不直接与目标系统交互隐蔽性好且能快速获得一批种子数据。DNS主动验证然后进行DNS枚举。DNS查询是主动的但相对HTTP更温和、更快。它可以发现没有Web服务的子域名。HTTP主动验证最后进行HTTP暴力枚举。这是最直接、但也是最“吵闹”和缓慢的方法。放在最后是因为我们已经通过前两步缩小了范围可以将发现的子域名也加入待检查列表并且它用于最终确认Web服务的存活状态。结果聚合每一步发现的结果都添加到一个总的集合set中利用集合的特性自动去重。5. 常见问题、优化与避坑指南在实际运行中你肯定会遇到各种各样的问题。下面是我在开发和测试过程中踩过的一些坑以及对应的解决方案。5.1 性能与稳定性问题问题1异步并发数设置过高导致本地端口耗尽或目标服务器拒绝连接。现象程序运行一段时间后出现大量ClientConnectorError或OSError: [Errno 24] Too many open files。排查Linux/Mac下可以用ulimit -n查看单进程文件描述符限制。每个TCP连接都会占用一个。解决限制并发数将max_concurrency信号量和连接器的limit设置为一个合理的值如50-200。可以通过参数传入方便调整。调整系统限制临时提高限制ulimit -n 65535仅当前会话有效。对于长期使用可能需要修改系统配置。使用更优的连接器aiohttp.TCPConnector(limit0)可以禁用连接限制但需谨慎使用。问题2目标网站有WAFWeb应用防火墙或速率限制导致IP被临时封禁。现象前期请求正常后期全部返回403/429状态码或连接超时。解决降低并发增加延迟在任务循环中使用asyncio.sleep()随机添加微小延迟如0.1-0.5秒模拟人类操作。使用代理池集成代理IP池在请求时随机切换代理。aiohttp支持通过proxy参数设置代理。设置User-Agent为ClientSession设置一个常见的、随机的User-Agent头避免使用明显的爬虫标识。问题3DNS查询时使用默认的本地DNS服务器可能慢或不稳定。解决为aiodns.DNSResolver()指定可靠的公共DNS服务器。resolver aiodns.DNSResolver(nameservers[8.8.8.8, 1.1.1.1, 114.114.114.114])5.2 数据准确性与完整性问题问题4从CT日志或搜索引擎获取的子域名包含大量无关结果。现象结果里出现了otherdomain.com或sub.example.com.cn这类不精确的匹配。解决加强过滤逻辑。最可靠的方法是检查域名是否以目标域名结尾。但要注意像example.com.cn和example.com的区别。更严谨的做法是使用正则表达式或domainlib这样的库进行解析和匹配。import re def is_subdomain(candidate, main_domain): # 确保 candidate 以 .main_domain 结尾且前面还有点或直接就是main_domain pattern r^(.*\.)? re.escape(main_domain) $ return bool(re.match(pattern, candidate))问题5HTTP存活检查误判。现象有些服务返回非200状态码如403, 404但服务是存在的有些CDN或WAF对所有不存在的子域名都返回200和一个默认页面。解决综合判断不要只依赖状态码。可以结合响应体长度、标题关键字、特定指纹如Server头、页面标题来判断。对比基线如果目标有一个统一的404页面可以事先访问一个肯定不存在的子域名如random123456.target.com获取其“基线”响应特征如标题、特定文本、哈希值然后在检查其他子域名时与之对比显著不同则可能存活。使用HEAD请求对于只需要判断存活的场景使用session.head()比session.get()更快因为它不下载响应体。5.3 工程化与扩展建议1. 配置化管理将API Key、默认字典路径、并发数、超时时间等写入一个配置文件如config.yaml或config.ini方便管理和分享避免硬编码。2. 结果丰富化除了保存子域名列表还可以将每个子域名的信息结构化保存如JSON格式包含发现来源CT/搜索/DNS/暴力、IP地址、HTTP状态码、标题、Server头、响应长度等。这为后续的资产梳理和漏洞扫描提供了丰富上下文。3. 模块插件化将每种探测方法如virustotal_api,shodan_api,dns_zone_transfer设计成独立的插件。主程序通过配置文件加载启用的插件。这样扩展新功能只需编写新插件无需修改核心代码。4. 分布式与持久化对于超大型目标如*.google.com单机资源可能不足。可以考虑将任务队列如待检查的子域名列表放入Redis由多个工作节点并发消费。结果也存入数据库如SQLite/MySQL便于查询和去重。5. 遵守法律法规与道德规范这是最重要的一点。仅将此工具用于你拥有书面授权测试的目标或你自己的资产。未经授权的扫描可能违反《计算机信息网络国际联网安全保护管理办法》等相关法律法规构成违法行为。务必在合法、合规的范围内使用技术。最后这个“子域名探测二”项目从一个简单的脚本演化成了一个具备一定工程化雏形的工具。它体现了在实际安全工作中如何将多个信息源、多种技术手段有机结合在效率、广度和深度之间寻找平衡。代码本身是一个起点你可以根据自己的需求轻松地添加新的模块如Fofa、Shodan的API查询优化过滤算法或者为其设计一个图形界面。希望这套代码和思路能成为你自动化信息收集工作流中一块坚实的基石。