Patched code sample

import asyncio
from aiohttp import web

# This set defines the trusted hostnames the server will respond to.
# For a server intended only for local access, this should be limited to
# loopback addresses and hostnames. This is the core of the DNS rebinding fix.
# Prior to the fix, this check was missing, allowing any 'Host' header.
ALLOWED_HOSTS = {"localhost", "127.0.0.1", "[::1]"}

@web.middleware
async def host_header_validation_middleware(request, handler):
    """
    AIOHTTP middleware to validate the Host header against an allowlist.

    This middleware represents the fix for CVE-2025-66416. It prevents DNS
    rebinding attacks by ensuring that the server only processes requests
    that are explicitly addressed to a trusted hostname.

    An attacker using DNS rebinding would trick a user's browser into sending a
    request to a local IP (e.g., 127.0.0.1) but with the 'Host' header of the
    attacker's domain. This middleware detects that mismatch and blocks the
    request, which was not done by default in vulnerable versions.
    """
    # aiohttp's `request.host` correctly provides the host part of the header.
    # We strip the port, if present, before checking against the allowlist.
    host = request.host.split(':')[0]
    
    if host not in ALLOWED_HOSTS:
        # If the host is not in the allowlist, deny the request with 403 Forbidden.
        # This is the security control that patches the vulnerability.
        return web.Response(
            text=f"Forbidden: Invalid 'Host' header '{request.host}'.",
            status=403
        )
    # If the host is valid, proceed to the actual request handler.
    return await handler(request)

async def handle_mcp_tool_invoke(request):
    """
    A mock handler for an MCP (Model Context Protocol) endpoint.

    In a real-world scenario, this could invoke a sensitive tool or access local
    resources. With the middleware in place, this handler is now protected from
    being invoked by malicious websites via DNS rebinding.
    """
    print(f"Received a valid request for {request.path} from host '{request.host}'")
    return web.json_response({
        "status": "success",
        "message": "MCP tool invoked successfully on the local server."
    })

async def main():
    """
    Sets up and runs the secure HTTP-based MCP server.
    """
    # The middleware is added to the application, ensuring it runs for all requests.
    # This represents the fixed version where DNS rebinding protection is enabled.
    app = web.Application(middlewares=[host_header_validation_middleware])
    
    # Add a route representing a resource exposed by the MCP server.
    app.add_routes([
        web.post('/mcp/invoke', handle_mcp_tool_invoke),
        web.get('/mcp/sse', handle_mcp_tool_invoke) # Example for SSE transport
    ])

    # Per the CVE, the vulnerable setup is a server running on localhost.
    runner = web.AppRunner(app)
    await runner.setup()
    site = web.TCPSite(runner, '127.0.0.1', 8080)
    
    print("Starting secure MCP server on http://127.0.0.1:8080...")
    print("FIX APPLIED: Host header validation is now ENABLED by default.")
    await site.start()

    # Keep the server running to allow for testing.
    while True:
        await asyncio.sleep(3600)

if __name__ == '__main__':
    # This runnable example demonstrates the fix.
    # To test the fix:
    # 1. Run this script.
    # 2. In a new terminal, execute the following curl commands.
    #
    # Test with a valid 'Host' header (this request will be ALLOWED):
    # curl -v -X POST http://127.0.0.1:8080/mcp/invoke -H "Host: localhost"
    #
    # Simulate a DNS rebinding attack with a malicious 'Host' header
    # (this request will be BLOCKED by the middleware with a 403 Forbidden):
    # curl -v -X POST http://127.0.0.1:8080/mcp/invoke -H "Host: malicious-site.com"

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