(原创)通过Python编写Xray批量测试脚本 仅用于学习和方法研究
简介： Xray 批量控制工具是一款功能强大的网络扫描工具，旨在帮助安全专业人员发现和评估网络应用程序中的漏洞。本工具结合了高度定制化的扫描功能和多线程处理，使用户能够高效地扫描多个目标并快速获取漏洞信息。在这篇介绍中，我们将深入了解Xray批量控制工具的主要特点和用途。
主要特点：
多线程扫描： Xray 批量控制工具支持多线程处理，最多可以同时扫描多个目标。这意味着用户可以高效地处理大量目标，减少扫描时间。
目标定制化： 用户可以轻松定义要扫描的目标，只需将目标列表保存在名为 'targets.txt' 的文件中即可。工具会自动读取这些目标并将其添加到扫描队列中。
结果输出： Xray会对扫描结果进行分析，并将结果输出到HTML文件中，便于用户查看漏洞详情。每个目标都会生成一个单独的HTML文件，以保持结果的整洁性。
漏洞检测： 工具会检测漏洞并在结果中标识出来。用户可以轻松识别漏洞并采取适当的措施来修复或改进网络应用程序的安全性。
操作系统兼容性： Xray 批量控制工具具有跨平台兼容性，可以在Windows和Linux操作系统上运行，使其适用于各种环境。
用途：
Xray 批量控制工具适用于多种安全场景，包括但不限于：
漏洞评估： 安全团队可以使用工具来主动评估其网络应用程序的漏洞，以确保其应用程序的安全性。
渗透测试： 渗透测试人员可以利用Xray来模拟潜在的攻击，识别漏洞并提供改进建议。
漏洞修复验证： 维护人员可以使用工具来验证已修复的漏洞，确保问题已彻底解决。
安全意识培训： 安全教育培训人员可以使用Xray来演示不安全的应用程序行为，以提高员工的安全意识。
总结：
Xray 批量控制工具是一款强大的网络扫描工具，为安全专业人员提供了有效评估和改进网络应用程序安全性的手段。其多线程处理、目标定制化和漏洞检测功能使其成为网络安全领域的重要工具。无论是用于漏洞评估、渗透测试还是安全意识培训，Xray批量控制工具都能够帮助用户更好地理解和维护其网络应用程序的安全性。
以上介绍来自本站自建的GPT服务，自打网站被黑之后，一直在研究如何加强网站的防御，以及如何对网站进行尽可能的测试，所以写了这一系列的文章，能够使得一些工具实现串成串的效果。
代码如下：
新版(1、支持跨平台操作2、线程数决定于CPU数量)：
# -*- coding: utf-8 -*-
import os
import re
import time
import sys
import platform
import multiprocessing
from queue import Queue
from threading import Thread
print('''
__ __ _____ ____ ____ _
| \/ |_ _| ___| _ \ / ___| / \ ___ _ __
| |\/| | | | | |_ | |_) | | _ / _ \ / __| '_ \
| | | | |_| | _| | __/| |_| |/ ___ \ | (__| | | |
|_| |_|\__, |_| |_| \____/_/ \_(_)___|_| |_|
|___/
Xray Batch Control tool {1.7.2#main}
*Internal version code sensitive and confidential*
''')
def readfile():
# 创建一个队列用于存储扫描目标
target_queue = Queue()
# 从文件 'targets.txt' 读取目标并加入队列
targets_list = [line.strip() for line in open('targets.txt', encoding='utf-8')]
for target in targets_list:
target_queue.put(target)
return target_queue
def format_time(seconds):
# 将秒数格式化为时:分:秒的形式
hours, remainder = divmod(int(seconds), 3600)
minutes, seconds = divmod(remainder, 60)
return f"{hours:02d}:{minutes:02d}:{seconds:02d}"
def scan(target_queue, xray_path, total_targets):
start_time = time.time()
while True:
try:
target = target_queue.get()
# 从目标中提取域名部分
domain_match = re.search(r'(.*?//)?(.*)', target)
if domain_match:
domain = domain_match.group(2)
# 替换域名中的特殊字符，用于生成输出文件名
domain = domain.replace(":", "_").replace("/", "_")
# 构建扫描命令
command = f"{xray_path} webscan --basic-crawler {target} --html-output ./output/{domain}.html"
output = os.popen(command)
processed_targets = total_targets - target_queue.qsize()
elapsed_time = time.time() - start_time
targets_per_second = processed_targets / elapsed_time
remaining_targets = target_queue.qsize()
estimated_time = remaining_targets / targets_per_second if targets_per_second > 0 else 0
print(f"[{time.strftime('%H:%M:%S', time.localtime(time.time()))}] [INFO] 正在处理目标：{processed_targets}/{total_targets}，估计还需要 {format_time(estimated_time)}。")
try:
result = output.read()
if "[Vuln:" in result:
print(f"[{time.strftime('%H:%M:%S', time.localtime(time.time()))}] [INFO] 发现漏洞：{result}")
except UnicodeDecodeError:
pass
finally:
target_queue.task_done()
def get_platform_info():
system = platform.system() # 获取操作系统名称
architecture = platform.architecture() # 获取架构和位数信息
if system == 'Linux':
# 在Linux上，可以使用os.uname()获取更详细的架构信息
uname = os.uname()
machine = uname.machine
if 'aarch64' in machine:
return f"Linux Arm64 {architecture[0]}"
elif 'arm' in machine:
return f"Linux Arm {architecture[0]}"
elif 'x86_64' in machine:
return f"Linux x86_64 {architecture[0]}"
elif 'i386' in machine:
return f"Linux i386 {architecture[0]}"
else:
return f"Linux {machine} {architecture[0]}"
elif system == 'Windows':
machine = platform.machine()
if machine == 'AMD64':
return f"Windows x64 {architecture[0]}"
elif machine == 'i386':
return f"Windows x86 {architecture[0]}"
else:
return f"Windows {machine} {architecture[0]}"
else:
return f"Unsupported OS: {system}"
def select_xray_executable():
platform_info = get_platform_info()
if "Linux" in platform_info:
if "Arm64" in platform_info:
return "xray_linux_arm64"
elif "i386" in platform_info:
return "xray_linux_386"
elif "x86_64" in platform_info:
return "xray_linux_amd64"
elif "Windows" in platform_info:
if "i386" in platform_info:
return "xray_windows_386.exe"
elif "x64" in platform_info:
return "xray_windows_amd64.exe"
elif "Darwin" in platform_info:
if "Arm64" in platform_info:
return "xray_darwin_arm64"
elif "x86_64" in platform_info:
return "xray_darwin_amd64"
return "Unsupported platform"
if __name__ == "__main__":
# 根据操作系统选择 xray 可执行文件路径
xray_path = select_xray_executable()
target_queue = readfile()
total_targets = target_queue.qsize()
# 使用适量的线程来并行扫描目标
usr_threads = multiprocessing.cpu_count() #线程数，可以固定修改为10等
for i in range(usr_threads):
t = Thread(target=scan, args=(target_queue, xray_path, total_targets))
t.daemon = True
t.start()
target_queue.join()
print(f"[{time.strftime('%H:%M:%S', time.localtime(time.time()))}] [INFO] 所有任务已完成！")历史版本：
# -*- coding: utf-8 -*-
import os
import re
import time
import sys
from queue import Queue
from threading import Thread
print('''
__ __ _____ ____ ____ _
| \/ |_ _| ___| _ \ / ___| / \ ___ _ __
| |\/| | | | | |_ | |_) | | _ / _ \ / __| '_ \
| | | | |_| | _| | __/| |_| |/ ___ \ | (__| | | |
|_| |_|\__, |_| |_| \____/_/ \_(_)___|_| |_|
|___/
Xray Batch Control tool {1.7.1#main}
*Internal version code sensitive and confidential*
''')
def readfile():
# 创建一个队列用于存储扫描目标
target_queue = Queue()
# 从文件 'targets.txt' 读取目标并加入队列
targets_list = [line.strip() for line in open('targets.txt', encoding='utf-8')]
for target in targets_list:
target_queue.put(target)
return target_queue
def format_time(seconds):
# 将秒数格式化为时:分:秒的形式
hours, remainder = divmod(int(seconds), 3600)
minutes, seconds = divmod(remainder, 60)
return f"{hours:02d}:{minutes:02d}:{seconds:02d}"
def scan(target_queue, xray_path, total_targets):
start_time = time.time()
while True:
try:
target = target_queue.get()
# 从目标中提取域名部分
domain_match = re.search(r'(.*?//)?(.*)', target)
if domain_match:
domain = domain_match.group(2)
# 替换域名中的特殊字符，用于生成输出文件名
domain = domain.replace(":", "_").replace("/", "_")
# 构建扫描命令
command = f"{xray_path} webscan --basic-crawler {target} --html-output ./output/{domain}.html"
output = os.popen(command)
processed_targets = total_targets - target_queue.qsize()
elapsed_time = time.time() - start_time
targets_per_second = processed_targets / elapsed_time
remaining_targets = target_queue.qsize()
estimated_time = remaining_targets / targets_per_second if targets_per_second > 0 else 0
print(f"[{time.strftime('%H:%M:%S', time.localtime(time.time()))}] [INFO] 正在处理目标：{processed_targets}/{total_targets}，估计还需要 {format_time(estimated_time)}。")
try:
result = output.read()
if "[Vuln:" in result:
print(f"[{time.strftime('%H:%M:%S', time.localtime(time.time()))}] [INFO] 发现漏洞：{result}")
except UnicodeDecodeError:
pass
finally:
target_queue.task_done()
if __name__ == "__main__":
# 根据操作系统选择 xray 可执行文件路径
xray_path = "xray_windows_amd64.exe" if sys.platform == 'win32' else "xray_linux_amd64"
target_queue = readfile()
total_targets = target_queue.qsize()
# 使用适量的线程来并行扫描目标
for i in range(10):
t = Thread(target=scan, args=(target_queue, xray_path, total_targets))
t.daemon = True
t.start()
target_queue.join()
print(f"[{time.strftime('%H:%M:%S', time.localtime(time.time()))}] [INFO] 所有任务已完成！")
