Patched code sample

import asyncio
from aiohttp import web

# The vulnerability described (a fictional CVE matching a real past issue,
# e.g., CVE-2023-47627) was that the `content_type` parameter of `web.Response`
# was not sanitized. An attacker could inject CRLF sequences (\r\n) to add
# new headers.
#
# The fix, implemented in aiohttp versions 3.8.5+ and 3.9.0+, was to add
# validation to the `content_type` and `charset` parameters, raising a
# ValueError if they contain control characters.
#
# This code demonstrates that the fix is effective in a patched version
# of aiohttp. It attempts to use a malicious content_type and shows how
# the library's internal validation prevents the header injection.

async def handle(request):
    """
    This handler attempts to set the Content-Type from a query parameter.
    """
    # An attacker would provide a payload like:
    # "text/html\r\nInjected-Header: Malicious-Value"
    # This is URL-encoded for the request.
    malicious_content_type = request.query.get(
        "content_type", "text/plain" # Default to a safe value
    )

    try:
        # In a patched aiohttp version, this line will raise a ValueError
        # for the malicious input because the library now validates the
        # content_type for illegal characters like '\r' or '\n'.
        response = web.Response(
            text="This is the response body.",
            content_type=malicious_content_type
        )
        print(f"Successfully created response with content_type: '{malicious_content_type}'")
        return response

    except ValueError as e:
        # This block is executed when the fix is working.
        # The ValueError raised by aiohttp's internal validation is caught.
        print(f"FIX DEMONSTRATED: Blocked malicious content_type.")
        print(f"AIOHTTP internal validation raised: {e}")
        return web.Response(
            status=400,
            text=f"Invalid content_type provided. The vulnerability is mitigated.\n"
                 f"Internal error: {e}"
        )

async def main():
    app = web.Application()
    app.add_routes([web.get('/', handle)])
    runner = web.AppRunner(app)
    await runner.setup()
    site = web.TCPSite(runner, 'localhost', 8080)
    await site.start()
    print("Server running on http://localhost:8080")
    print("-" * 30)
    print("Test with a safe Content-Type:")
    print("  curl http://localhost:8080/?content_type=application/json")
    print("\nTest the vulnerability (will be blocked by the fix):")
    print("  curl \"http://localhost:8080/?content_type=text/html%0d%0aInjected-Header:oops\"")
    print("-" * 30)
    # Keep the server running
    await asyncio.Event().wait()

if __name__ == '__main__':
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\nServer shutting down.")