diff --git a/tests/server/test_dns_rebinding_fix.py b/tests/server/test_dns_rebinding_fix.py new file mode 100644 index 0000000..83a3d3a --- /dev/null +++ b/tests/server/test_dns_rebinding_fix.py @@ -0,0 +1,174 @@ +""" +Test that DNS rebinding protection is properly disabled for containerized deployments. + +This test verifies that the fix for MCP 1.23.x DNS rebinding protection works correctly. +Without the fix, requests with Host headers that don't match the default allowed list +(127.0.0.1:*, localhost:*, [::1]:*) would be rejected with a 421 Misdirected Request error. +""" + +import httpx +import pytest + + +@pytest.mark.integration +async def test_accepts_various_host_headers(): + """Test that the MCP server accepts requests with various Host headers. + + This test simulates what happens in containerized deployments where the Host + header might be a k8s service DNS name, a proxied hostname, or other values + that don't match the default allowed list. + + Without the DNS rebinding protection fix, these requests would fail with: + - 421 Misdirected Request (for Host header mismatch) + - 403 Forbidden (for Origin header mismatch) + """ + mcp_url = "http://localhost:8000/mcp" + + # Test various Host headers that would be rejected by DNS rebinding protection + test_cases = [ + { + "name": "Kubernetes service DNS", + "headers": { + "Host": "nextcloud-mcp-server.default.svc.cluster.local:8000", + "Content-Type": "application/json", + "Accept": "application/json, text/event-stream", + }, + }, + { + "name": "Custom domain", + "headers": { + "Host": "mcp.example.com:8000", + "Content-Type": "application/json", + "Accept": "application/json, text/event-stream", + }, + }, + { + "name": "Proxied hostname", + "headers": { + "Host": "proxy.internal:8000", + "Content-Type": "application/json", + "Accept": "application/json, text/event-stream", + }, + }, + { + "name": "Default localhost (should always work)", + "headers": { + "Host": "localhost:8000", + "Content-Type": "application/json", + "Accept": "application/json, text/event-stream", + }, + }, + ] + + # Create a simple initialize request payload + initialize_request = { + "jsonrpc": "2.0", + "method": "initialize", + "params": { + "protocolVersion": "2024-11-05", + "capabilities": {}, + "clientInfo": {"name": "test-client", "version": "1.0.0"}, + }, + "id": 1, + } + + async with httpx.AsyncClient() as client: + for test_case in test_cases: + print(f"\n๐Ÿงช Testing: {test_case['name']}") + print(f" Host header: {test_case['headers']['Host']}") + + response = await client.post( + mcp_url, + json=initialize_request, + headers=test_case["headers"], + timeout=10.0, + ) + + # With DNS rebinding protection enabled (MCP 1.23 default), these would fail with: + # - 421 Misdirected Request (Host header not in allowed list) + # - 403 Forbidden (Origin header not in allowed list) + # + # With our fix (enable_dns_rebinding_protection=False), they should succeed + assert response.status_code in [200, 202], ( + f"Request failed for {test_case['name']}: " + f"status={response.status_code}, " + f"headers={test_case['headers']}, " + f"body={response.text[:200]}" + ) + + print(f" โœ… Status: {response.status_code}") + + # For SSE responses (status 200), verify we got SSE format + # For JSON responses (status 202), verify we got valid JSON + if response.status_code == 200: + # SSE response - should start with "event: message" or similar + response_text = response.text + assert "event:" in response_text or "data:" in response_text, ( + f"Expected SSE format for {test_case['name']}, got: {response_text[:200]}" + ) + print(" โœ… Received SSE stream response") + elif response.status_code == 202: + # JSON response for notifications + response_json = response.json() + assert "jsonrpc" in response_json or response_json is None, ( + f"Invalid response for {test_case['name']}: {response_json}" + ) + print(" โœ… Received JSON response") + + +@pytest.mark.integration +async def test_dns_rebinding_protection_is_disabled(): + """Verify that DNS rebinding protection is actually disabled in the configuration. + + This test makes a request that would DEFINITELY fail if DNS rebinding protection + was enabled with default settings (only allowing 127.0.0.1:*, localhost:*, [::1]:*). + """ + mcp_url = "http://localhost:8000/mcp" + + # Use a Host header that would NEVER be in the default allowed list + malicious_host = "evil.attacker.com:8000" + + initialize_request = { + "jsonrpc": "2.0", + "method": "initialize", + "params": { + "protocolVersion": "2024-11-05", + "capabilities": {}, + "clientInfo": {"name": "test-client", "version": "1.0.0"}, + }, + "id": 1, + } + + async with httpx.AsyncClient() as client: + response = await client.post( + mcp_url, + json=initialize_request, + headers={ + "Host": malicious_host, + "Content-Type": "application/json", + "Accept": "application/json, text/event-stream", + }, + timeout=10.0, + ) + + # If DNS rebinding protection was enabled, this would return: + # - 421 Misdirected Request (Host header validation failed) + # + # Since we disabled it, this should succeed (status 200 or 202) + assert response.status_code in [200, 202], ( + f"DNS rebinding protection may still be enabled! " + f"Request with Host='{malicious_host}' was rejected: " + f"status={response.status_code}, body={response.text[:500]}" + ) + + # Verify we got a valid response (SSE or JSON) + if response.status_code == 200: + response_text = response.text + assert "event:" in response_text or "data:" in response_text, ( + f"Expected SSE format, got: {response_text[:200]}" + ) + + print("โœ… DNS rebinding protection is properly disabled") + print( + f" Request with Host '{malicious_host}' succeeded: {response.status_code}" + )