Python 實戰：內網滲透中的信息收集自動化腳本（2）

用途限制聲明，本文僅用于網絡安全技術研究、教育與知識分享。文中涉及的滲透測試方法與工具，嚴禁用于未經授權的網絡攻擊、數據竊取或任何違法活動。任何因不當使用本文內容導致的法律后果，作者及發布平臺不承擔任何責任。滲透測試涉及復雜技術操作，可能對目標系統造成數據損壞、服務中斷等風險。讀者需充分評估技術能力與潛在后果，在合法合規前提下謹慎實踐。

這次主要介紹子域名爆破腳本，關于這個腳本大家應該不會陌生一個工具，oneforall子域名爆破，接下來我們就從代碼的角度看看是如何進行子域名爆破的,先提供源碼，再進行逐個模塊分析

import dns.resolver
import socket
import argparse
import time
from datetime import datetime
from concurrent.futures import ThreadPoolExecutor, as_completed
import logging
from typing import List, Optional# 配置日志輸出
logging.basicConfig(level=logging.INFO,format='%(asctime)s - %(levelname)s - %(message)s',handlers=[logging.FileHandler("dns_exploration.log"),logging.StreamHandler()]
)
logger = logging.getLogger(__name__)def reverse_dns(ip: str) -> Optional[List[str]]:"""反向DNS查詢，獲取IP對應的域名信息"""try:result = socket.gethostbyaddr(ip)return [result[0]] + result[1]except (socket.herror, socket.timeout):return Nonedef dns_request(domain: str, record_type: str = 'A') -> List[str]:"""DNS查詢函數，支持多種記錄類型:param domain: 待查詢域名:param record_type: DNS記錄類型(A, AAAA, CNAME, MX, NS等):return: 查詢到的記錄列表"""ips = []try:resolver = dns.resolver.Resolver()resolver.timeout = 5  # 超時時間resolver.lifetime = 10  # 總生命周期result = resolver.resolve(domain, record_type)for answer in result:record_value = answer.to_text()ips.append(record_value)logger.info(f"域名: {domain} | 類型: {record_type} | 記錄值: {record_value}")# 對A/AAAA記錄進行反向查詢if record_type in ['A', 'AAAA']:reverse_domains = reverse_dns(record_value)if reverse_domains:logger.info(f"  反向解析: {', '.join(reverse_domains)}")except dns.resolver.NXDOMAIN:logger.debug(f"域名不存在: {domain} (類型: {record_type})")except dns.resolver.NoAnswer:logger.debug(f"無{record_type}記錄: {domain}")except (dns.exception.Timeout, dns.resolver.NoNameservers):logger.warning(f"查詢超時/無可用DNS服務器: {domain} (類型: {record_type})")except Exception as e:logger.error(f"查詢錯誤 {domain}: {str(e)}")return ipsdef check_subdomain(word: str, domain: str, record_types: List[str], nums: bool) -> List[str]:"""檢查單個子域名（含數字后綴變種）"""successes = []# 基礎子域名subdomain = f"{word}.{domain}"for rtype in record_types:if dns_request(subdomain, rtype):successes.append(subdomain)# 帶數字后綴的子域名if nums:for i in range(10):num_subdomain = f"{word}{i}.{domain}"for rtype in record_types:if dns_request(num_subdomain, rtype):successes.append(num_subdomain)return successesdef subdomain_search(domain: str, dictionary: List[str], nums: bool = False,record_types: List[str] = ['A'], threads: int = 5) -> List[str]:"""子域名枚舉主函數:param domain: 主域名:param dictionary: 子域名字典列表:param nums: 是否檢查數字后綴:param record_types: 需要查詢的DNS記錄類型:param threads: 線程數，控制并發:return: 所有存在的子域名列表"""successes = []total = len(dictionary)start_time = time.time()with ThreadPoolExecutor(max_workers=threads) as executor:# 提交所有任務futures = {executor.submit(check_subdomain, word, domain, record_types, nums): wordfor word in dictionary}# 跟蹤進度completed = 0for future in as_completed(futures):completed += 1progress = (completed / total) * 100logger.info(f"進度: {progress:.2f}% ({completed}/{total})")successes.extend(future.result())logger.info(f"枚舉完成，耗時: {time.time() - start_time:.2f}秒")return list(set(successes))  # 去重def save_results(results: List[str], domain: str) -> None:"""保存結果到文件"""timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")filename = f"subdomains_{domain}_{timestamp}.txt"with open(filename, "w") as f:for subdomain in sorted(results):f.write(f"{subdomain}\n")logger.info(f"結果已保存到: {filename}")def main():# 命令行參數解析parser = argparse.ArgumentParser(description="高級DNS子域名枚舉工具")parser.add_argument("-d", "--domain", required=True, help="目標主域名 (例如: google.com)")parser.add_argument("-f", "--file", default="subdomains.txt", help="子域名字典文件路徑")parser.add_argument("-n", "--nums", action="store_true", help="檢查帶數字(0-9)后綴的子域名")parser.add_argument("-t", "--types", default="A", help="DNS記錄類型，多個用逗號分隔 (例如: A,AAAA,CNAME)")parser.add_argument("-th", "--threads", type=int, default=5, help="并發線程數 (默認: 5)")args = parser.parse_args()# 解析記錄類型record_types = [t.strip().upper() for t in args.types.split(',')]# 讀取字典文件try:with open(args.file, "r", encoding="utf-8") as f:dictionary = [line.strip() for line in f if line.strip()]logger.info(f"已加載字典文件: {args.file} (共{len(dictionary)}個條目)")except FileNotFoundError:logger.error(f"字典文件不存在: {args.file}")return# 執行子域名枚舉logger.info(f"開始枚舉 {args.domain} 的子域名...")results = subdomain_search(domain=args.domain,dictionary=dictionary,nums=args.nums,record_types=record_types,threads=args.threads)# 輸出結果if results:logger.info(f"共發現 {len(results)} 個有效子域名:")for sub in sorted(results):logger.info(f"  - {sub}")save_results(results, args.domain)else:logger.info("未發現有效子域名")if __name__ == "__main__":main()

1. 模塊導入與日志配置

import dns.resolver
import socket
import argparse
import time
from datetime import datetime
from concurrent.futures import ThreadPoolExecutor, as_completed
import logging
from typing import List, Optional# 配置日志輸出
logging.basicConfig(level=logging.INFO,format='%(asctime)s - %(levelname)s - %(message)s',handlers=[logging.FileHandler("dns_exploration.log"),logging.StreamHandler()]
)
logger = logging.getLogger(__name__)

解析：

模塊選擇邏輯：
- dns.resolver：核心 DNS 查詢工具，用于解析域名的各類 DNS 記錄（替代原始版本的基礎查詢）。
- socket：僅保留反向 DNS 查詢功能（IP→域名映射）。
- argparse：實現命令行參數解析，讓工具支持靈活配置（如指定域名、字典文件等）。
- concurrent.futures：提供線程池支持，實現并發查詢以提升效率（解決原始版本單線程慢的問題）。
- logging：替代原始的print輸出，支持多級別日志（INFO/WARNING/ERROR）和雙目標輸出（控制臺 + 文件）。
- typing：添加類型注解，提升代碼可讀性和 IDE 支持（如List[str]表示字符串列表）。
日志配置細節：
- level=logging.INFO：默認輸出 INFO 及以上級別日志（DEBUG 信息需手動開啟）。
- format：包含時間戳、日志級別和消息內容，便于追溯執行過程。
- handlers：同時輸出到文件（dns_exploration.log）和控制臺，既保留記錄又方便實時查看。

2. 反向 DNS 查詢函數?`reverse_dns`

def reverse_dns(ip: str) -> Optional[List[str]]:"""反向DNS查詢，獲取IP對應的域名信息"""try:result = socket.gethostbyaddr(ip)return [result[0]] + result[1]except (socket.herror, socket.timeout):return None

解析：

功能定位：通過 IP 地址反向解析對應的域名（IP→域名映射），補充正向 DNS 查詢的信息。
參數與返回值：
- 輸入ip: str：待查詢的 IP 地址（IPv4 或 IPv6）。
- 返回Optional[List[str]]：可能為None（解析失敗）或域名列表（主域名 + 別名）。
異常處理：
- socket.herror：主機不存在或無反向記錄時觸發。
- socket.timeout：查詢超時（新增處理，解決原始版本未處理超時的問題）。
實現細節：socket.gethostbyaddr(ip)返回元組(主機名, 別名列表, IP列表)，此處提取主機名和別名合并為列表返回。

3. DNS 查詢函數?`dns_request`

def dns_request(domain: str, record_type: str = 'A') -> List[str]:"""DNS查詢函數，支持多種記錄類型:param domain: 待查詢域名:param record_type: DNS記錄類型(A, AAAA, CNAME, MX, NS等):return: 查詢到的記錄列表"""ips = []try:resolver = dns.resolver.Resolver()resolver.timeout = 5  # 超時時間resolver.lifetime = 10  # 總生命周期result = resolver.resolve(domain, record_type)for answer in result:record_value = answer.to_text()ips.append(record_value)logger.info(f"域名: {domain} | 類型: {record_type} | 記錄值: {record_value}")# 對A/AAAA記錄進行反向查詢if record_type in ['A', 'AAAA']:reverse_domains = reverse_dns(record_value)if reverse_domains:logger.info(f"  反向解析: {', '.join(reverse_domains)}")except dns.resolver.NXDOMAIN:logger.debug(f"域名不存在: {domain} (類型: {record_type})")except dns.resolver.NoAnswer:logger.debug(f"無{record_type}記錄: {domain}")except (dns.exception.Timeout, dns.resolver.NoNameservers):logger.warning(f"查詢超時/無可用DNS服務器: {domain} (類型: {record_type})")except Exception as e:logger.error(f"查詢錯誤 {domain}: {str(e)}")return ips

解析：

核心改進：從原始版本僅支持 A 記錄查詢，擴展為支持任意 DNS 記錄類型（A/AAAA/CNAME/MX 等）。
** resolver 配置 **：
- timeout=5：單次查詢超時時間（5 秒），避免長時間阻塞。
- lifetime=10：總查詢生命周期（10 秒），包含重試時間，防止無限等待。
記錄處理邏輯：
- 對查詢結果result遍歷，將每條記錄轉為文本（如 A 記錄的 IP、CNAME 的域名）并存儲。
- 僅對 A/AAAA 記錄（IP 地址記錄）觸發反向查詢，因為其他類型（如 MX）的記錄值不是 IP，無需反向解析。
異常處理增強：
- NXDOMAIN：域名不存在（DEBUG 級別，避免日志冗余）。
- NoAnswer：域名存在但無指定類型記錄（如查詢 AAAA 但域名僅支持 IPv4）。
- Timeout/NoNameservers：警告級別，提示網絡或 DNS 服務器問題。
- 通用異常捕獲：防止未預料的錯誤導致程序崩潰。

4. 子域名檢查函數?`check_subdomain`

def check_subdomain(word: str, domain: str, record_types: List[str], nums: bool) -> List[str]:"""檢查單個子域名（含數字后綴變種）"""successes = []# 基礎子域名subdomain = f"{word}.{domain}"for rtype in record_types:if dns_request(subdomain, rtype):successes.append(subdomain)# 帶數字后綴的子域名if nums:for i in range(10):num_subdomain = f"{word}{i}.{domain}"for rtype in record_types:if dns_request(num_subdomain, rtype):successes.append(num_subdomain)return successes

解析：

功能定位：處理單個字典詞的子域名生成與查詢，是并發任務的最小執行單元。
邏輯拆分：
1. 基礎子域名：將字典詞與主域名拼接（如word + .google.com），查詢所有指定記錄類型。
2. 數字后綴子域名：若nums=True，在字典詞后拼接 0-9 數字（如word0.google.com），再查詢記錄類型。
去重預備：返回的successes列表可能包含重復子域名（如同一子域名同時有 A 和 CNAME 記錄），后續在主函數中通過list(set(successes))去重。

5. 子域名枚舉主函數?`subdomain_search`

def subdomain_search(domain: str, dictionary: List[str], nums: bool = False,record_types: List[str] = ['A'], threads: int = 5) -> List[str]:"""子域名枚舉主函數:param domain: 主域名:param dictionary: 子域名字典列表:param nums: 是否檢查數字后綴:param record_types: 需要查詢的DNS記錄類型:param threads: 線程數，控制并發:return: 所有存在的子域名列表"""successes = []total = len(dictionary)start_time = time.time()with ThreadPoolExecutor(max_workers=threads) as executor:# 提交所有任務futures = {executor.submit(check_subdomain, word, domain, record_types, nums): wordfor word in dictionary}# 跟蹤進度completed = 0for future in as_completed(futures):completed += 1progress = (completed / total) * 100logger.info(f"進度: {progress:.2f}% ({completed}/{total})")successes.extend(future.result())logger.info(f"枚舉完成，耗時: {time.time() - start_time:.2f}秒")return list(set(successes))  # 去重

解析：

并發核心：使用ThreadPoolExecutor實現多線程查詢，解決原始版本單線程效率低下的問題。max_workers=threads控制并發數（默認 5，可通過命令行調整）。
任務管理：
- executor.submit：為每個字典詞提交一個check_subdomain任務，返回Future對象（代表異步任務）。
- as_completed(futures)：迭代已完成的任務，實時獲取結果并更新進度。
進度跟蹤：通過completed/total計算進度百分比，讓用戶了解枚舉進展（原始版本無進度提示）。
性能優化：
- 并發查詢大幅縮短總耗時（尤其字典較大時）。
- 最終通過list(set(successes))去重，避免同一子域名被多次記錄。

6. 結果保存與主函數?`main`

def save_results(results: List[str], domain: str) -> None:"""保存結果到文件"""timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")filename = f"subdomains_{domain}_{timestamp}.txt"with open(filename, "w") as f:for subdomain in sorted(results):f.write(f"{subdomain}\n")logger.info(f"結果已保存到: {filename}")def main():# 命令行參數解析parser = argparse.ArgumentParser(description="高級DNS子域名枚舉工具")parser.add_argument("-d", "--domain", required=True, help="目標主域名 (例如: google.com)")parser.add_argument("-f", "--file", default="subdomains.txt", help="子域名字典文件路徑")parser.add_argument("-n", "--nums", action="store_true", help="檢查帶數字(0-9)后綴的子域名")parser.add_argument("-t", "--types", default="A", help="DNS記錄類型，多個用逗號分隔 (例如: A,AAAA,CNAME)")parser.add_argument("-th", "--threads", type=int, default=5, help="并發線程數 (默認: 5)")args = parser.parse_args()# 解析記錄類型record_types = [t.strip().upper() for t in args.types.split(',')]# 讀取字典文件try:with open(args.file, "r", encoding="utf-8") as f:dictionary = [line.strip() for line in f if line.strip()]logger.info(f"已加載字典文件: {args.file} (共{len(dictionary)}個條目)")except FileNotFoundError:logger.error(f"字典文件不存在: {args.file}")return# 執行子域名枚舉logger.info(f"開始枚舉 {args.domain} 的子域名...")results = subdomain_search(domain=args.domain,dictionary=dictionary,nums=args.nums,record_types=record_types,threads=args.threads)# 輸出結果if results:logger.info(f"共發現 {len(results)} 個有效子域名:")for sub in sorted(results):logger.info(f"  - {sub}")save_results(results, args.domain)else:logger.info("未發現有效子域名")

解析：

save_results?函數：
- 生成帶時間戳的文件名（如subdomains_google.com_20240520_153045.txt），避免結果文件覆蓋。
- 對結果排序后寫入，方便閱讀和后續處理。
main?函數（程序入口）：
1. 參數解析：通過argparse定義命令行參數，支持用戶靈活配置目標域名、字典文件、記錄類型等（原始版本需硬編碼修改參數）。
2. 字典加載：讀取子域名字典文件，過濾空行，統計條目數（便于進度計算）。
3. 流程調度：調用subdomain_search執行枚舉，最終輸出結果并保存到文件。