content: """Tests for StreamableHTTP server DNS rebinding protection."""
  
  import logging
  import multiprocessing
  import socket
  from collections.abc import AsyncGenerator
  from contextlib import asynccontextmanager
  
  import httpx
  import pytest
  import uvicorn
  from starlette.applications import Starlette
  from starlette.routing import Mount
  from starlette.types import Receive, Scope, Send
  
  from mcp.server import Server
  from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
  from mcp.server.transport_security import TransportSecuritySettings
  from mcp.types import Tool
  from tests.test_helpers import wait_for_server
  
  logger = logging.getLogger(__name__)
  SERVER_NAME = "test_streamable_http_security_server"
  
  
  @pytest.fixture
  def server_port() -> int:
      with socket.socket() as s:
          s.bind(("127.0.0.1", 0))
          return s.getsockname()[1]
  
  
  @pytest.fixture
  def server_url(server_port: int) -> str:  # pragma: no cover
      return f"http://127.0.0.1:{server_port}"
  
  
  class SecurityTestServer(Server):  # pragma: no cover
      def __init__(self):
          super().__init__(SERVER_NAME)
  
      async def on_list_tools(self) -> list[Tool]:
          return []
  
  
  def run_server_with_settings(port: int, security_settings: TransportSecuritySettings | None = None):  # pragma: no cover
      """Run the StreamableHTTP server with specified security settings."""
      app = SecurityTestServer()
  
      # Create session manager with security settings
      session_manager = StreamableHTTPSessionManager(
          app=app,
          json_response=False,
          stateless=False,
          security_settings=security_settings,
      )
  
      # Create the ASGI handler
      async def handle_streamable_http(scope: Scope, receive: Receive, send: Send) -> None:
          await session_manager.handle_request(scope, receive, send)
  
      # Create Starlette app with lifespan
      @asynccontextmanager
      async def lifespan(app: Starlette) -> AsyncGenerator[None, None]:
          async with session_manager.run():
              yield
  
      routes = [
          Mount("/", app=handle_streamable_http),
      ]
  
      starlette_app = Starlette(routes=routes, lifespan=lifespan)
      uvicorn.run(starlette_app, host="127.0.0.1", port=port, log_level="error")
  
  
  def start_server_process(port: int, security_settings: TransportSecuritySettings | None = None):
      """Start server in a separate process."""
      process = multiprocessing.Process(target=run_server_with_settings, args=(port, security_settings))
      process.start()
      # Wait for server to be ready to accept connections
      wait_for_server(port)
      return process
  
  
  @pytest.mark.anyio
  async def test_streamable_http_security_default_settings(server_port: int):
      """Test StreamableHTTP with default security settings (protection enabled)."""
      process = start_server_process(server_port)
  
      try:
          # Test with valid localhost headers
          async with httpx.AsyncClient(timeout=5.0) as client:
              # POST request to initialize session
              response = await client.post(
                  f"http://127.0.0.1:{server_port}/",
                  json={"jsonrpc": "2.0", "method": "initialize", "id": 1, "params": {}},
                  headers={
                      "Accept": "application/json, text/event-stream",
                      "Content-Type": "application/json",
                  },
              )
              assert response.status_code == 200
              assert "mcp-session-id" in response.headers
  
      finally:
          process.terminate()
          process.join()
  
  
  @pytest.mark.anyio
  async def test_streamable_http_security_invalid_host_header(server_port: int):
      """Test StreamableHTTP with invalid Host header."""
      security_settings = TransportSecuritySettings(enable_dns_rebinding_protection=True)
      process = start_server_process(server_port, security_settings)
  
      try:
          # Test with invalid host header
          headers = {
              "Host": "evil.com",
              "Accept": "application/json, text/event-stream",
              "Content-Type": "application/json",
          }
  
          async with httpx.AsyncClient(timeout=5.0) as client:
              response = await client.post(
                  f"http://127.0.0.1:{server_port}/",
                  json={"jsonrpc": "2.0", "method": "initialize", "id": 1, "params": {}},
                  headers=headers,
              )
              assert response.status_code == 421
              assert response.text == "Invalid Host header"
  
      finally:
          process.terminate()
          process.join()
  
  
  @pytest.mark.anyio
  async def test_streamable_http_security_invalid_origin_header(server_port: int):
      """Test StreamableHTTP with invalid Origin header."""
      security_settings = TransportSecuritySettings(enable_dns_rebinding_protection=True, allowed_hosts=["127.0.0.1:*"])
      process = start_server_process(server_port, security_settings)
  
      try:
          # Test with invalid origin header
          headers = {
              "Origin": "http://evil.com",
              "Accept": "application/json, text/event-stream",
              "Content-Type": "application/json",
          }
  
          async with httpx.AsyncClient(timeout=5.0) as client:
              response = await client.post(
                  f"http://127.0.0.1:{server_port}/",
                  json={"jsonrpc": "2.0", "method": "initialize", "id": 1, "params": {}},
                  headers=headers,
              )
              assert response.status_code == 403
              assert response.text == "Invalid Origin header"
  
      finally:
          process.terminate()
          process.join()
  
  
  @pytest.mark.anyio
  async def test_streamable_http_security_invalid_content_type(server_port: int):
      """Test StreamableHTTP POST with invalid Content-Type header."""
      process = start_server_process(server_port)
  
      try:
          async with httpx.AsyncClient(timeout=5.0) as client:
              # Test POST with invalid content type
              response = await client.post(
                  f"http://127.0.0.1:{server_port}/",
                  headers={
                      "Content-Type": "text/plain",
                      "Accept": "application/json, text/event-stream",
                  },
                  content="test",
              )
              assert response.status_code == 400
              assert response.text == "Invalid Content-Type header"
  
              # Test POST with missing content type
              response = await client.post(
                  f"http://127.0.0.1:{server_port}/",
                  headers={"Accept": "application/json, text/event-stream"},
                  content="test",
              )
              assert response.status_code == 400
              assert response.text == "Invalid Content-Type header"
  
      finally:
          process.terminate()
          process.join()
  
  
  @pytest.mark.anyio
  async def test_streamable_http_security_disabled(server_port: int):
      """Test StreamableHTTP with security disabled."""
      settings = TransportSecuritySettings(enable_dns_rebinding_protection=False)
      process = start_server_process(server_port, settings)
  
      try:
          # Test with invalid host header - should still work
          headers = {
              "Host": "evil.com",
              "Accept": "application/json, text/event-stream",
              "Content-Type": "application/json",
          }
  
          async with httpx.AsyncClient(timeout=5.0) as client:
              response = await client.post(
                  f"http://127.0.0.1:{server_port}/",
                  json={"jsonrpc": "2.0", "method": "initialize", "id": 1, "params": {}},
                  headers=headers,
              )
              # Should connect successfully even with invalid host
              assert response.status_code == 200
  
      finally:
          process.terminate()
          process.join()
  
  
  @pytest.mark.anyio
  async def test_streamable_http_security_custom_allowed_hosts(server_port: int):
      """Test StreamableHTTP with custom allowed hosts."""
      settings = TransportSecuritySettings(
          enable_dns_rebinding_protection=True,
          allowed_hosts=["localhost", "127.0.0.1", "custom.host"],
          allowed_origins=["http://localhost", "http://127.0.0.1", "http://custom.host"],
      )
      process = start_server_process(server_port, settings)
  
      try:
          # Test with custom allowed host
          headers = {
              "Host": "custom.host",
              "Accept": "application/json, text/event-stream",
              "Content-Type": "application/json",
          }
  
          async with httpx.AsyncClient(timeout=5.0) as client:
              response = await client.post(
                  f"http://127.0.0.1:{server_port}/",
                  json={"jsonrpc": "2.0", "method": "initialize", "id": 1, "params": {}},
                  headers=headers,
              )
              # Should connect successfully with custom host
              assert response.status_code == 200
      finally:
          process.terminate()
          process.join()
  
  
  @pytest.mark.anyio
  async def test_streamable_http_security_get_request(server_port: int):
      """Test StreamableHTTP GET request with security."""
      security_settings = TransportSecuritySettings(enable_dns_rebinding_protection=True, allowed_hosts=["127.0.0.1"])
      process = start_server_process(server_port, security_settings)
  
      try:
          # Test GET request with invalid host header
          headers = {
              "Host": "evil.com",
              "Accept": "text/event-stream",
          }
  
          async with httpx.AsyncClient(timeout=5.0) as client:
              response = await client.get(f"http://127.0.0.1:{server_port}/", headers=headers)
              assert response.status_code == 421
              assert response.text == "Invalid Host header"
  
          # Test GET request with valid host header
          headers = {
              "Host": "127.0.0.1",
              "Accept": "text/event-stream",
          }
  
          async with httpx.AsyncClient(timeout=5.0) as client:
              # GET requests need a session ID in StreamableHTTP
              # So it will fail with "Missing session ID" not security error
              response = await client.get(f"http://127.0.0.1:{server_port}/", headers=headers)
              # This should pass security but fail on session validation
              assert response.status_code == 400
              body = response.json()
              assert "Missing session ID" in body["error"]["message"]
  
      finally:
          process.terminate()
          process.join()
  