test(frontend): align SSRF e2e with sanitized errors

tests/frontend/test_http_status_propagation.py

@@ -17,7 +17,11 @@
 import requests
 
 from tests.utils.managed_process import DynamoFrontendProcess, ManagedProcess
-from tests.utils.network_canary import ConnectionCanary, running_canary
+from tests.utils.network_canary import (
+    ConnectionCanary,
+    assert_blocked_url_refused,
+    running_canary,
+)
 from tests.utils.port_utils import ServicePorts
 
 MODEL_NAME = "test-http-status-prop"
@@ -138,10 +142,7 @@ def test_python_backend_ssrf_rejection_is_4xx_with_zero_egress(
         timeout=30,
     )
 
-    assert 400 <= response.status_code < 500, response.text
-    assert response.json()["message"] == "Invalid request", response.text
-    assert "is in a blocked range" not in response.text, response.text
-    outbound_canary.assert_no_connection()
+    assert_blocked_url_refused(response, outbound_canary, "is in a blocked range")
 
 
 @pytest.mark.timeout(30)

tests/mm_router/test_vllm_frontend_ssrf_e2e.py

@@ -20,7 +20,11 @@
 from tests.mm_router.utils import COMMON_PROCESS_KWARGS, build_vllm_gpu_mem_args
 from tests.utils.http_checks import check_health_ready, model_registered
 from tests.utils.managed_process import ManagedProcess
-from tests.utils.network_canary import ConnectionCanary, running_canary
+from tests.utils.network_canary import (
+    ConnectionCanary,
+    assert_blocked_url_refused,
+    running_canary,
+)
 from tests.utils.port_utils import reserved_ports
 
 VLLM_MM_MODEL = os.getenv("DYN_TEST_VLLM_MM_MODEL", "Qwen/Qwen3-VL-2B-Instruct")
@@ -168,13 +172,8 @@ def test_blocked_url_is_refused_and_never_fetched(
         timeout=180,
     )
 
-    assert 400 <= response.status_code < 500, (
-        f"[{topology}] expected a 4xx for a blocked media URL, got "
-        f"HTTP {response.status_code}: {response.text[:2000]}"
-    )
-    expected_detail = {
+    private_detail = {
         "vllm_processor": "blocked range",
         "rust_decoding": "Direct IP access is not allowed",
     }[topology]
-    assert expected_detail in response.text, response.text
-    canary.assert_no_connection()
+    assert_blocked_url_refused(response, canary, private_detail, label=topology)

tests/utils/network_canary.py

@@ -9,7 +9,10 @@
 import socketserver
 import threading
 from contextlib import contextmanager
-from typing import Iterator
+from typing import TYPE_CHECKING, Iterator
+
+if TYPE_CHECKING:
+    import requests
 
 
 class _Handler(socketserver.BaseRequestHandler):
@@ -60,6 +63,24 @@ def assert_no_connection(self, timeout: float = 0.25) -> None:
             )
 
 
+def assert_blocked_url_refused(
+    response: requests.Response,
+    canary: ConnectionCanary,
+    private_detail: str,
+    label: str = "",
+) -> None:
+    """Assert a 4xx that hides the policy detail and never reached the canary."""
+    prefix = f"[{label}] " if label else ""
+    assert 400 <= response.status_code < 500, (
+        f"{prefix}expected a 4xx for a blocked media URL, got "
+        f"HTTP {response.status_code}: {response.text[:2000]}"
+    )
+    # The frontend returns only the sanitized class message.
+    assert response.json()["message"] == "Invalid request", response.text
+    assert private_detail not in response.text, response.text
+    canary.assert_no_connection()
+
+
 @contextmanager
 def running_canary() -> Iterator[ConnectionCanary]:
     server = ConnectionCanary()