1. 项目概述：为什么我们要自己动手写一个目录扫描器？

在网络安全测试或者日常的网站资产梳理中，目录扫描是一个基础但至关重要的环节。市面上有像 dirsearch、gobuster 这样的成熟工具，功能强大，开箱即用。那为什么我们还要“从零开始”用 Python 写一个呢？这就像学开车，直接给你一辆自动挡的顶级跑车，你当然能开，但你可能永远不知道离合器、变速箱是怎么配合工作的。自己动手写一个目录扫描器，核心目的不是要造一个比它们更好的轮子，而是为了深入理解这个“轮子”的构造原理、轴承怎么转、辐条怎么受力。

通过这个项目，你将彻底搞懂几个关键问题：HTTP 请求在代码层面是如何发送和接收的？多线程/多协程是如何大幅提升扫描效率的？如何处理网络请求中的超时、重定向、异常状态？如何设计一个灵活可扩展的字典管理和结果过滤机制？这些知识点，是你看十篇工具使用教程也未必能内化的。当你亲手实现一遍，以后再使用任何扫描工具，你都能一眼看穿它的核心逻辑，甚至能根据特定场景快速定制或修改。这对于想从事安全研发、自动化测试，或者单纯想提升 Python 工程能力的开发者来说，是一次绝佳的练手机会。

2. 核心需求与设计思路拆解

2.1 核心功能定义

一个最基本的目录扫描器，需要完成以下核心功能：

1. 字典加载 ：从一个文本文件中读取可能的目录或文件路径（如 /admin , /backup.zip , /config.php 等），构成待扫描的队列。
2. 目标拼接与请求 ：将字典中的每一项与目标网站的基础 URL 进行拼接，形成完整的 URL，然后发起 HTTP 请求。
3. 响应分析 ：根据 HTTP 响应状态码（如 200、403、404、500）、响应内容长度、标题等特征，判断该路径是否存在或可访问。
4. 结果输出 ：将有效的发现（例如状态码为 200、302 的路径）以清晰的方式输出到屏幕或文件。
5. 并发控制 ：为了提高扫描速度，必须支持同时发起多个 HTTP 请求，这就需要引入多线程、多进程或异步 IO 机制。

2.2 技术方案选型与考量

为什么用 Python？因为它拥有极其丰富和易用的网络库（如 requests , aiohttp ），并发编程库（ threading , asyncio ），以及强大的标准库支持，能让开发者更专注于逻辑本身而非底层细节。

• HTTP 客户端 ：初期学习和快速原型验证，首选 requests 库。它语法简洁直观，错误处理完善，适合理解基础流程。当我们追求极致性能时，可以升级为异步版本的 aiohttp 。
• 并发模型 ：
  • 多线程 ( threading ) ：易于理解，适合 I/O 密集型任务（网络请求正是此类）。但 Python 的全局解释器锁（GIL）会限制 CPU 密集型任务的并行，不过对我们的扫描任务影响不大。需要注意线程间的资源竞争（如共享的结果列表、计数器）。
  • 线程池 ( concurrent.futures.ThreadPoolExecutor ) ：比直接操作 threading 模块更高级、更安全，它帮我们管理了线程的生命周期和任务队列，是推荐的选择。
  • 异步 IO ( asyncio + aiohttp ) ：性能最高，资源消耗最小，是生产级别工具的选择。但学习曲线稍陡，涉及 async/await 语法和事件循环概念。
• 字典管理 ：一个好的扫描器应支持从文件加载字典，并能方便地添加自定义条目或进行动态规则生成（如根据响应内容生成新的猜测）。
• 结果过滤 ：需要能过滤掉常见的“误报”，比如虽然返回 200 状态码，但内容却是统一的“404 Not Found”页面（通过内容长度或关键字匹配来识别）。

基于“从零学习”的目标，我们将采用 requests + ThreadPoolExecutor 的方案，在保证性能的同时，让代码逻辑足够清晰，便于理解和调试。

3. 基础版本实现：单线程扫描器

让我们先搭建一个最基础的、没有并发功能的扫描器，把核心流程跑通。

3.1 环境准备与依赖安装

首先，确保你的 Python 环境（建议 3.6 以上）已经就绪。我们需要安装 requests 库。

pip install requests

3.2 核心代码逐步实现

我们创建一个名为 simple_scanner.py 的文件。

第一步：导入库和定义基础参数

import requests
import sys
from urllib.parse import urljoin

class SimpleDirScanner:
    def __init__(self, target_url, wordlist_path):
        """
        初始化扫描器
        :param target_url: 目标网站基础URL，如 http://example.com
        :param wordlist_path: 字典文件路径
        """
        self.target_url = target_url.rstrip('/')  # 去除末尾可能存在的斜杠
        self.wordlist_path = wordlist_path
        self.results = []  # 存储扫描结果
        # 设置一个会话对象，可以复用 TCP 连接，提升效率
        self.session = requests.Session()
        # 设置一个合理的超时时间，避免长时间等待
        self.timeout = 5
        # 自定义请求头，有些网站会校验 User-Agent
        self.headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
        }

这里使用 requests.Session() 是一个小技巧。相比每次请求都新建连接，Session 可以保持一个连接池，复用 TCP 连接，对于需要发送大量请求到同一主机的场景，能显著减少开销。

第二步：加载字典文件

    def load_wordlist(self):
        """从文件加载字典，并做基础清理"""
        try:
            with open(self.wordlist_path, 'r', encoding='utf-8', errors='ignore') as f:
                # 读取每一行，去除两端空白字符，过滤掉空行和注释行（以#开头）
                words = [line.strip() for line in f if line.strip() and not line.strip().startswith('#')]
            print(f"[*] 成功加载字典，共 {len(words)} 个条目。")
            return words
        except FileNotFoundError:
            print(f"[!] 错误：字典文件 '{self.wordlist_path}' 未找到。")
            sys.exit(1)
        except Exception as e:
            print(f"[!] 加载字典时发生未知错误：{e}")
            sys.exit(1)

注意 errors='ignore' 参数，它能处理字典文件中可能存在的非 UTF-8 编码字符，避免程序因编码问题意外崩溃。过滤空行和注释是提高字典质量的好习惯。

第三步：发送请求与判断逻辑 这是最核心的函数。

    def probe_path(self, path):
        """探测单个路径"""
        # 拼接完整URL
        full_url = urljoin(self.target_url + '/', path.lstrip('/'))
        try:
            # 发送GET请求，带上自定义头部和超时设置
            response = self.session.get(full_url, headers=self.headers, timeout=self.timeout, allow_redirects=False)
            
            # 基础判断逻辑：状态码为 2xx (成功) 或 3xx (重定向) 通常表示资源存在
            if response.status_code in [200, 201, 204, 301, 302, 307, 308]:
                # 记录结果：URL，状态码，响应内容长度
                result = {
                    'url': full_url,
                    'status': response.status_code,
                    'length': len(response.content)
                }
                self.results.append(result)
                # 实时输出发现
                print(f"[+] {response.status_code:3d} - {len(response.content):7d} - {full_url}")
            # 403 状态码也值得关注，它表示路径存在但禁止访问
            elif response.status_code == 403:
                print(f"[*] 403 - {len(response.content):7d} - {full_url} (禁止访问)")
                
        except requests.exceptions.Timeout:
            print(f"[!] 超时 - {full_url}")
        except requests.exceptions.ConnectionError:
            print(f"[!] 连接错误 - {full_url}，可能目标不可达或路径非法。")
        except requests.exceptions.RequestException as e:
            print(f"[!] 请求异常 - {full_url}: {e}")

这里有几个关键点：

1. urljoin 函数能智能地处理 URL 拼接，避免出现 http://a.com//admin 这样的双斜杠错误。
2. allow_redirects=False 非常重要。如果我们允许自动重定向，那么扫描 /admin 时，如果它跳转到 /login.php ，我们收到的状态码将是 200（登录页面的），这会导致误判。设置为 False 后，我们收到的是 302 状态码，能更准确地知道 /admin 这个端点本身的行为。
3. 我们不仅关注状态码，还记录了响应内容长度。这对于后续的智能过滤至关重要（比如，10 个不同的路径都返回长度相同的 200 页面，那很可能是统一的错误页）。

第四步：主循环与结果输出

    def run(self):
        """启动扫描"""
        words = self.load_wordlist()
        print(f"[*] 开始扫描目标：{self.target_url}")
        print("[*] 格式：[状态码] - [内容长度] - [URL]")
        print("-" * 60)
        
        for i, word in enumerate(words, 1):
            self.probe_path(word)
            # 简单进度提示，每扫描100个路径打印一次
            if i % 100 == 0:
                print(f"[*] 进度：{i}/{len(words)}")
        
        print("-" * 60)
        print(f"[*] 扫描结束。共发现 {len(self.results)} 个潜在有效路径。")
        # 可以将结果保存到文件
        if self.results:
            with open('scan_results.txt', 'w') as f:
                for r in self.results:
                    f.write(f"{r['status']} - {r['length']} - {r['url']}\n")
            print("[*] 结果已保存至 scan_results.txt")

if __name__ == '__main__':
    # 简单的命令行参数处理
    if len(sys.argv) != 3:
        print("用法: python simple_scanner.py <目标URL> <字典文件>")
        print("示例: python simple_scanner.py http://testphp.vulnweb.com common.txt")
        sys.exit(1)
    
    target = sys.argv[1]
    wordlist = sys.argv[2]
    
    scanner = SimpleDirScanner(target, wordlist)
    scanner.run()

现在，你可以找一个测试网站（例如 http://testphp.vulnweb.com ）和一个常见的字典文件（如 common.txt ），运行 python simple_scanner.py http://testphp.vulnweb.com common.txt 来体验基础扫描。

实操心得 1：关于测试目标 千万不要在未经授权的情况下扫描任何非你所有的线上网站！这是违法的，且极不道德。可以使用专门为安全测试搭建的靶场，如 DVWA、bWAPP，或上面提到的 testphp.vulnweb.com 。自己用 Python 的 http.server 模块快速搭建一个本地服务器来测试也是好方法： python -m http.server 8080 ，然后扫描 http://localhost:8080 。

4. 性能飞跃：引入多线程并发扫描

基础版本是顺序执行的，扫描 1000 个路径就要发起 1000 次网络请求，大部分时间都在等待网络响应，效率极低。接下来我们引入线程池进行改造。

4.1 使用 ThreadPoolExecutor 进行重构

我们创建新文件 threaded_scanner.py ，复用大部分基础代码，主要修改 run 方法。

import concurrent.futures
# ... 保留之前的导入和 SimpleDirScanner 类定义，但我们将类名改为 ThreadedDirScanner

class ThreadedDirScanner(SimpleDirScanner):
    def __init__(self, target_url, wordlist_path, max_workers=20):
        """
        初始化多线程扫描器
        :param max_workers: 线程池最大工作线程数
        """
        super().__init__(target_url, wordlist_path)
        self.max_workers = max_workers
        # 用于线程安全地操作结果列表
        import threading
        self.lock = threading.Lock()

    def probe_path(self, path):
        """重写 probe_path 方法，注意线程安全"""
        full_url = urljoin(self.target_url + '/', path.lstrip('/'))
        try:
            response = self.session.get(full_url, headers=self.headers, timeout=self.timeout, allow_redirects=False)
            
            if response.status_code in [200, 201, 204, 301, 302, 307, 308]:
                result = {
                    'url': full_url,
                    'status': response.status_code,
                    'length': len(response.content)
                }
                # 使用锁来确保同时只有一个线程操作结果列表
                with self.lock:
                    self.results.append(result)
                print(f"[+] {response.status_code:3d} - {len(response.content):7d} - {full_url}")
            elif response.status_code == 403:
                print(f"[*] 403 - {len(response.content):7d} - {full_url} (禁止访问)")
                
        except requests.exceptions.Timeout:
            print(f"[!] 超时 - {full_url}")
        except requests.exceptions.ConnectionError:
            # 连接错误可能很频繁，可以选择静默或统一记录
            pass
        except requests.exceptions.RequestException:
            pass

    def run(self):
        words = self.load_wordlist()
        print(f"[*] 开始多线程扫描目标：{self.target_url} (线程数：{self.max_workers})")
        print("[*] 格式：[状态码] - [内容长度] - [URL]")
        print("-" * 60)
        
        # 使用线程池映射执行
        with concurrent.futures.ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            # executor.map 会按顺序将words中的每个元素传递给probe_path，但执行是并发的
            # 它返回一个迭代器，我们通过list()来消费它，等待所有任务完成
            list(executor.map(self.probe_path, words))
        
        print("-" * 60)
        print(f"[*] 扫描结束。共发现 {len(self.results)} 个潜在有效路径。")
        if self.results:
            with open('threaded_scan_results.txt', 'w') as f:
                for r in self.results:
                    f.write(f"{r['status']} - {r['length']} - {r['url']}\n")
            print("[*] 结果已保存至 threaded_scan_results.txt")

if __name__ == '__main__':
    if len(sys.argv) not in [3, 4]:
        print("用法: python threaded_scanner.py <目标URL> <字典文件> [线程数]")
        print("示例: python threaded_scanner.py http://testphp.vulnweb.com common.txt 30")
        sys.exit(1)
    
    target = sys.argv[1]
    wordlist = sys.argv[2]
    workers = int(sys.argv[3]) if len(sys.argv) == 4 else 20
    
    scanner = ThreadedDirScanner(target, wordlist, max_workers=workers)
    scanner.run()

4.2 关键改进与原理

1. 线程池 ( ThreadPoolExecutor ) ：它管理着一个固定数量的工作线程。我们将 words 列表中的每个路径提交给线程池，线程池会自动分配空闲线程去执行 probe_path 函数。这避免了频繁创建和销毁线程的开销。
2. 线程安全 ( threading.Lock ) ：当多个线程同时尝试向 self.results 列表追加数据时，可能会发生数据错乱。 Lock 锁机制确保同一时间只有一个线程能执行 with self.lock: 块内的代码，从而安全地修改共享资源。
3. 异常处理简化 ：在多线程环境下，如果每个连接错误都打印，会导致控制台输出混乱。我们选择静默处理常见的网络异常，只输出成功或值得关注的发现。你也可以选择将错误记录到一个单独的日志文件中。
4. executor.map ：这是一个非常方便的函数，它接收一个函数和一个可迭代对象，并发地对每个元素应用该函数，并返回一个结果迭代器。我们用 list() 来迭代它，主要是为了等待所有 future 完成（因为 map 返回的是 Future 对象）， _ 表示我们不关心它的返回值（我们的结果存在 self.results 里）。

实操心得 2：线程数的设置 max_workers 并非越大越好。线程数过多会导致大量的线程上下文切换，反而降低性能，并且会给目标服务器带来巨大压力，可能触发对方的防护机制（如 IP 封禁）。一般建议设置在 20-50 之间，具体取决于你的网络环境和目标服务器的承受能力。可以先从 20 开始测试。

5. 功能增强：智能过滤与结果去重

现在我们的扫描器能跑了，但结果中可能包含大量“误报”。比如，目标网站对所有不存在的路径都返回一个自定义的 200 状态码“404页面”，我们的扫描器就会把它们全部记录下来。我们需要引入智能过滤。

5.1 基于内容长度和关键字的过滤

我们新增一个 Filter 类，并在扫描器中集成它。

class ResponseFilter:
    def __init__(self):
        # 存储已知的“错误页面”特征
        self.error_lengths = set()  # 错误页面的内容长度集合
        self.error_keywords = []    # 错误页面包含的关键字列表
        
    def learn_from_404(self, response):
        """从一个明确的404响应中学习特征"""
        self.error_lengths.add(len(response.content))
        # 可以提取页面标题或特定文本作为关键字，这里简单示例
        # 实际中可以更复杂，如计算页面相似度
        if \"404\" in response.text or \"Not Found\" in response.text:
            self.error_keywords.append(\"404\")
            self.error_keywords.append(\"Not Found\")
    
    def is_error_page(self, response):
        """判断一个响应是否是错误页面"""
        # 规则1：内容长度匹配已知的错误页面长度
        if len(response.content) in self.error_lengths:
            return True
        # 规则2：响应内容包含已知的错误关键字
        content_text = response.text.lower()
        for keyword in self.error_keywords:
            if keyword.lower() in content_text:
                return True
        return False

class SmartDirScanner(ThreadedDirScanner):
    def __init__(self, target_url, wordlist_path, max_workers=20):
        super().__init__(target_url, wordlist_path, max_workers)
        self.filter = ResponseFilter()
        # 首先探测几个肯定不存在的随机路径，学习错误页面特征
        self.learn_error_pattern()
    
    def learn_error_pattern(self):
        """学习目标站点的错误页面模式"""
        print(\"[*] 正在学习目标站点错误页面特征...\")
        test_paths = [f\"wp-admin-{random.randint(10000,99999)}\", f\"api-test-{random.randint(10000,99999)}\"]
        for path in test_paths:
            url = urljoin(self.target_url + '/', path)
            try:
                resp = self.session.get(url, headers=self.headers, timeout=3, allow_redirects=False)
                if resp.status_code >= 400: # 404, 403 等都可以作为学习样本
                    self.filter.learn_from_404(resp)
            except:
                pass
        print(f\"[*] 学习完成。已记录 {len(self.filter.error_lengths)} 个错误页面长度特征。\")
    
    def probe_path(self, path):
        """重写 probe_path，加入过滤逻辑"""
        full_url = urljoin(self.target_url + '/', path.lstrip('/'))
        try:
            response = self.session.get(full_url, headers=self.headers, timeout=self.timeout, allow_redirects=False)
            
            # 新增过滤判断
            if self.filter.is_error_page(response):
                # 被识别为错误页面，静默跳过
                return
                
            if response.status_code in [200, 201, 204, 301, 302, 307, 308]:
                result = {
                    'url': full_url,
                    'status': response.status_code,
                    'length': len(response.content)
                }
                with self.lock:
                    self.results.append(result)
                print(f\"[+] {response.status_code:3d} - {len(response.content):7d} - {full_url}\")
            elif response.status_code == 403:
                print(f\"[*] 403 - {len(response.content):7d} - {full_url} (禁止访问)\")
                
        except requests.exceptions.Timeout:
            pass
        except requests.exceptions.ConnectionError:
            pass
        except requests.exceptions.RequestException:
            pass

5.2 结果去重与排序

扫描完成后，我们可能发现多个路径指向同一个实际页面（比如 /admin 和 /admin/ 可能都返回 200）。我们可以对结果进行简单的去重和排序，让报告更清晰。可以在 run 方法的最后，保存结果前添加：

    def run(self):
        # ... 前面的扫描代码 ...
        print(f\"[*] 扫描结束。原始发现 {len(self.results)} 个路径。\")
        
        # 去重：根据 (状态码, 内容长度, URL) 去重，更精确的去重可以只根据URL
        unique_results = []
        seen = set()
        for r in self.results:
            # 创建一个唯一标识符，这里用状态码+长度+URL
            identifier = (r['status'], r['length'], r['url'])
            if identifier not in seen:
                seen.add(identifier)
                unique_results.append(r)
        
        # 排序：按状态码和URL排序
        unique_results.sort(key=lambda x: (x['status'], x['url']))
        
        print(f\"[*] 去重后剩余 {len(unique_results)} 个唯一路径。\")
        
        if unique_results:
            with open('smart_scan_results.txt', 'w') as f:
                for r in unique_results:
                    f.write(f\"{r['status']} - {r['length']} - {r['url']}\\n\")
            print(\"[*] 结果已保存至 smart_scan_results.txt\")

注意事项：过滤策略的权衡 智能过滤是一把双刃剑。过于激进的过滤可能会漏掉一些真正有价值的发现，比如一个设计独特的 404 页面可能没有被我们的学习样本捕捉到。因此，在生产环境中，通常会采用“宽进严出”的策略：扫描时记录所有非 404 状态码的响应，在后期分析阶段再通过更复杂的算法（如相似度比对）进行过滤。我们当前实现的是一种简易的实时过滤，适合快速扫描和初步判断。

6. 进阶优化：支持更多协议与扩展性设计

一个成熟的扫描器不会只支持 HTTP GET 方法。我们还可以考虑以下扩展方向，这些可以作为你后续迭代的练习：

6.1 支持多种 HTTP 方法

有些接口或管理后台可能只响应 POST、PUT 等请求。我们可以扩展字典格式和请求逻辑。

• 字典格式 ：可以约定在字典条目中使用前缀，如 GET:/api/user , POST:/login.php 。
• 请求分发 ：在 probe_path 函数中解析方法，然后调用 session.post() , session.put() 等。

6.2 递归扫描

发现一个目录（如 /admin ）后，可以自动将其作为新的基础路径，加载字典进行下一层扫描（如 /admin/config , /admin/backup ）。这需要维护一个待扫描队列，并注意避免循环扫描和深度控制。

6.3 报告生成

将结果输出为 HTML、JSON 或 Markdown 格式的报告，包含更丰富的信息，如响应头部、页面标题截图（需要集成浏览器引擎）等。

6.4 配置化管理

将线程数、超时时间、过滤规则、请求头等参数抽取到配置文件（如 YAML、JSON）或命令行参数中，使工具更灵活。

6.5 性能监控与速率限制

添加扫描进度条（如使用 tqdm 库）、实时显示扫描速度（请求数/秒），并实现可配置的请求速率限制（ time.sleep ），以避免对目标造成过大压力。

7. 常见问题与排查技巧实录

在实际编写和运行过程中，你肯定会遇到各种问题。这里记录一些典型问题和解决方法。

问题 1：程序运行后立刻崩溃，报错 ModuleNotFoundError: No module named 'requests'

• 原因 ： requests 库没有安装。
• 解决 ：在命令行中执行 pip install requests 。如果使用虚拟环境，请确保在正确的环境中安装。

问题 2：扫描速度非常慢，甚至不如单线程。

• 原因 ：
  1. 线程数 ( max_workers ) 设置过高，导致大量线程竞争 CPU 和网络资源，上下文切换开销巨大。
  2. 目标服务器响应慢或网络延迟高。
  3. 字典文件过大，内存加载慢。
• 排查与解决 ：
  1. 降低线程数 ：尝试将 max_workers 设置为 10、20、30 分别测试，找到性能拐点。
  2. 调整超时时间 ：适当降低 timeout （如设为 3 秒），让慢请求更快失败，释放线程。
  3. 检查网络 ：使用 ping 或 traceroute 检查到目标服务器的网络状况。
  4. 优化字典 ：使用更精简、更相关的字典。可以边读边扫，而不是一次性加载全部到内存（使用生成器）。

问题 3：扫描结果中出现了大量 403 状态码，但没有 200。

• 原因 ：目标服务器配置了严格的访问控制，禁止目录浏览，并对大多数路径返回 403 Forbidden。
• 分析 ：403 状态码本身就是一个重要发现，它表明路径是存在的，只是当前无权访问。这可能意味着需要身份认证（如 cookies, tokens）或存在其他绕过方式。
• 应对 ：在扫描器中集成 Cookie 或自定义 Header 支持，模拟已登录状态再进行扫描。

问题 4：程序运行一段时间后，控制台不再输出，但也没有结束，像是“卡住”了。

• 原因 ：可能是遇到了一个长时间无响应的请求，线程被挂起；或者发生了未处理的异常导致某个线程崩溃，但线程池还在等待。
• 排查 ：
  1. 添加更详细的日志 ：在异常捕获块中打印更多信息，尤其是 RequestException 。
  2. 使用调试器 ：在 IDE 中设置断点，查看线程状态。
  3. 检查目标 ：手动用浏览器访问几个字典中的路径，看是否正常响应。
• 解决 ：确保 timeout 参数已设置，并考虑为整个扫描任务设置一个总超时。

问题 5：扫描自己本地搭建的服务器，发现所有路径都返回 200。

• 原因 ：很多简单的 HTTP 服务器（如 Python 的 http.server ）对于不存在的文件，会返回一个列出目录内容的页面（状态码 200），而不是 404。
• 解决 ：这正是我们实现“智能过滤”的意义所在。通过先学习错误页面特征（ learn_error_pattern ），可以过滤掉这些统一的 200 响应。对于更复杂的情况，可能需要结合响应内容相似度算法。

通过这个从零构建 Python 目录扫描器的过程，我们不仅学会了如何使用 requests 和并发编程，更重要的是理解了网络扫描工具背后的核心思想： 高效地发送请求、智能地分析响应、安全地处理异常 。你可以以此为基础，不断添加新功能，比如集成更强大的异步框架 aiohttp ，添加递归扫描、子域名枚举模块，甚至设计一个图形化界面。记住，工具是死的，思路是活的。理解原理，你就能创造出适应各种场景的解决方案。