Patched code sample

import asyncio
from aiohttp import web

# The fictional CVE-2026-34519 describes a real vulnerability type:
# HTTP Header Injection via the 'reason' phrase.
# The actual, similar vulnerability was CVE-2023-47627, fixed in aiohttp 3.9.0.
#
# This code demonstrates the logic of the fix. The fix involves validating
# the 'reason' string to ensure it does not contain carriage return (CR) or
# line feed (LF) characters, which could be used to inject new lines and headers.

def represents_patched_response_creation(status, reason, text):
    """
    This function simulates the creation of an HTTP response in a patched
    version of aiohttp. It includes the crucial validation step on the 'reason' phrase.
    """
    # --- THE FIX ---
    # Before using the reason phrase, validate it to prevent CRLF injection.
    # If a forbidden character is found, an exception is raised.
    if '\r' in reason or '\n' in reason:
        raise ValueError("Invalid characters in reason phrase (CR/LF detected)")
    # --- END OF FIX ---

    # If validation passes, construct the response as usual.
    # The actual aiohttp.web.Response object handles this internally.
    return web.Response(status=status, reason=reason, text=text)


async def handle_request(request):
    """
    A web handler to demonstrate the effect of the fix.
    It takes a 'reason' from the query parameters to simulate user-controlled input.
    """
    # Simulate an attacker providing a malicious reason phrase via a query parameter.
    user_provided_reason = request.query.get("reason", "OK")

    print(f"\nAttempting to create response with reason: '{user_provided_reason!r}'")

    try:
        # Use the patched creation logic, which contains the security check.
        response = represents_patched_response_creation(
            status=200,
            reason=user_provided_reason,
            text="Response created successfully."
        )
        print("SUCCESS: Response created safely.")
        return response
    except ValueError as e:
        # The malicious attempt is caught, and an error response is returned.
        error_message = f"FAILURE: Vulnerability prevented. {e}"
        print(error_message)
        return web.Response(status=400, text=error_message)


async def main():
    """Sets up and runs the web server."""
    app = web.Application()
    app.add_routes([web.get('/', handle_request)])
    runner = web.AppRunner(app)
    await runner.setup()
    site = web.TCPSite(runner, 'localhost', 8080)
    await site.start()

    print("=" * 60)
    print("Server started on http://localhost:8080")
    print("This code demonstrates the fix for a header injection vulnerability.")
    print("\nTo test, use a client like curl in separate terminal windows:")
    print("\n1. Test a valid request (no injection):")
    print("   curl -i http://localhost:8080?reason=All-Good")
    print("\n2. Test a malicious request (attempts injection):")
    print("   curl -i \"http://localhost:8080?reason=OK%0d%0aInjected-Header:%20evil\"")
    print("=" * 60)

    # Keep server running to allow for testing.
    await asyncio.Event().wait()


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