如何为Web服务器暴力目录扫描脚本应用多线程提速？

我写了一个Web服务器暴力目录扫描脚本，想通过多线程技术提升扫描速度，但自己尝试了多种方案都没成功，而且对Python线程技术不太熟悉。以下是我的脚本代码：

#! /usr/bin/python

#Usage: 
#   python3 script.py [URL] [WORDLIST]
import sys
import socket
import requests

try:
    rhost = sys.argv[1]
    wordlist = sys.argv[2]

    print ('[*] Checking RHOST... '),
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    try:
        status = s.connect_ex((rhost, 80))
        s.close()
        if status == 0:
            print ('[DONE]')
            pass
        else:
            print ('[FAIL]')
            print ('[!] Error: Cannot Reach RHOST %s\n' %(rhost))
            sys.exit(1)
    except socket.error:
        print ('[FAIL]')
        print ('[!] Error: Cannot Reach RHOST: %s\n' %(rhost))
        sys.exit(1)

    print ('[*] Parsing Wordlist... '),
    try:
        with open(wordlist) as file:
            to_check = file.read().strip().split('\n')
        print ('[DONE]')
        print ('[*] Total Paths to Check: %s' %(str(len(to_check))))
    except IOError:
        print ('[FAIL]')
        print ('[!] Error: Failed to Read Specified File\n')
        sys.exit(1)
    
    def checkpath(path):
        try:
            response = requests.get('http://' + rhost + '/' + path).status_code
        except Exception:
            print ('[!] Error: An Unexpected Error Occured')
            sys.exit(1)
        if response == 200:
            print ('[*] Valid Path Found: /%s' %(path))
    
    print ('\n[*] Beginning Scan...\n')
    for i in range(len(to_check)):
        checkpath(to_check[i])
    print ('\n[*] Scan Complete!')
except KeyboardInterrupt:
    print ('\n[!] Error: User Interrupted Scan')
    sys.exit(1)

多线程改造方案 #

网络请求是扫描速度的核心瓶颈，用多线程让多个请求并行发送，能大幅提升扫描效率。推荐使用Python标准库的concurrent.futures.ThreadPoolExecutor实现，它可以自动管理线程池，无需手动处理线程的创建和销毁，对新手友好。

修改后的多线程版本代码 #

#! /usr/bin/python

#Usage: 
#   python3 script.py [URL] [WORDLIST]
import sys
import socket
import requests
from concurrent.futures import ThreadPoolExecutor

# 可调整线程数，根据目标服务器和自身网络设置，建议10-50之间
THREAD_COUNT = 10

try:
    rhost = sys.argv[1]
    wordlist = sys.argv[2]

    print('[*] Checking RHOST... ', end='')
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    try:
        status = s.connect_ex((rhost, 80))
        s.close()
        if status == 0:
            print('[DONE]')
        else:
            print('[FAIL]')
            print(f'[!] Error: Cannot Reach RHOST {rhost}\n')
            sys.exit(1)
    except socket.error:
        print('[FAIL]')
        print(f'[!] Error: Cannot Reach RHOST: {rhost}\n')
        sys.exit(1)

    print('[*] Parsing Wordlist... ', end='')
    try:
        with open(wordlist) as file:
            # 过滤词表中的空行，避免无效请求
            to_check = [line.strip() for line in file if line.strip()]
        print('[DONE]')
        print(f'[*] Total Paths to Check: {len(to_check)}')
    except IOError:
        print('[FAIL]')
        print('[!] Error: Failed to Read Specified File\n')
        sys.exit(1)
    
    def checkpath(path):
        try:
            url = f'http://{rhost}/{path}'
            # 添加超时设置，防止单个请求卡住整个扫描
            response = requests.get(url, timeout=5)
            if response.status_code == 200:
                print(f'[*] Valid Path Found: /{path}')
        except requests.exceptions.RequestException as e:
            # 子线程中不直接退出程序，仅打印错误信息
            print(f'[!] Request Error for /{path}: {str(e)}')
    
    print('\n[*] Beginning Scan...\n')
    # 启动线程池，自动分配任务执行
    with ThreadPoolExecutor(max_workers=THREAD_COUNT) as executor:
        executor.map(checkpath, to_check)
    print('\n[*] Scan Complete!')
except KeyboardInterrupt:
    print('\n[!] Error: User Interrupted Scan')
    sys.exit(1)

关键改动说明 #

• 导入ThreadPoolExecutor：Python3.2+自带的线程池工具，简化多线程实现流程
• 设置THREAD_COUNT：可根据实际情况调整线程数量，过高可能触发目标反爬或本地网络拥堵
• 优化路径列表：过滤词表空行，减少无效请求
• 添加请求超时：防止单个请求卡住导致扫描停滞
• 调整异常处理：子线程遇错误仅打印信息，不终止整个程序
• 替换循环逻辑：用executor.map自动将路径分配给线程池，实现并行扫描

内容的提问来源于stack exchange，提问作者Kobra3390