Vulnerable code sample

import time

# This code is a conceptual representation of the vulnerability described in
# CVE-2026-48045 (a likely typo for CVE-2023-48045) in the zeroconf library
# prior to version 0.149.12. It is not the original source code but is
# designed to functionally demonstrate the described flaws.

# --- Mock objects to simulate the environment ---

class MockDNSMessage:
    """Represents an incoming DNS message."""
    _MAX_MSG_ABSOLUTE = 8966

    def __init__(self, data, is_truncated=False):
        if len(data) > self._MAX_MSG_ABSOLUTE:
            raise ValueError("Message too large")
        self.data = data
        self.is_truncated = is_truncated

class MockTimer:
    """Represents a timer object."""
    def __init__(self, callback):
        self._callback = callback
        # In a real implementation, this would involve event loops.

    def cancel(self):
        # In a real implementation, this would cancel the scheduled callback.
        pass

# --- Vulnerable Class Representation ---

class VulnerableAsyncListener:
    """
    A simplified representation of zeroconf's AsyncListener class
    before the fix for CVE-2026-48045.
    """
    def __init__(self):
        # Vulnerability: Unbounded dictionaries to store deferred queries and timers.
        self._deferred = {}
        self._timers = {}

    def _handle_deferred(self, addr):
        """Placeholder for the callback that handles deferred queries."""
        # In the real code, this would process and clear the deferred items.
        # The vulnerability lies in the accumulation of items before this runs.
        if addr in self._timers:
            del self._timers[addr]
        if addr in self._deferred:
            # Process and clear
            del self._deferred[addr]

    def handle_query_or_defer(self, msg: MockDNSMessage, addr: tuple):
        """
        This method contains the logic vulnerable to memory and CPU exhaustion.
        It retains all truncated queries without any limits.
        """
        # The vulnerability is triggered by messages with the TC (Truncation) bit set.
        if msg.is_truncated:
            # VULNERABILITY 1: Unbounded list growth per source address.
            # Every truncated query is appended to a list associated with its source address.
            # There is no check on the length of this list.
            if addr not in self._deferred:
                self._deferred[addr] = []
            self._deferred[addr].append(msg)

            # VULNERABILITY 2: Unbounded timer creation.
            # A new timer is armed for each new source address, leading to an
            # unbounded number of timer objects.
            if addr in self._timers:
                self._timers[addr].cancel()
            self._timers[addr] = MockTimer(callback=lambda: self._handle_deferred(addr))

            # VULNERABILITY 3: Potential for quadratic CPU burn.
            # If any subsequent processing on deferred items iterates over the
            # list, the work done grows quadratically as more items are added
            # for the same address. We simulate a simple form of this work.
            work_counter = 0
            for _ in self._deferred.get(addr, []):
                # Simulate a trivial CPU-bound operation.
                work_counter += 1
            # In a real attack, total processing time for N packets from one address
            # would be proportional to 1+2+3+...+N, i.e., O(N^2).

        else:
            # Non-truncated messages would be handled by other logic.
            pass