fix(cli): check the Host header on every local API server request (v1) (#6789)

src/google/adk/cli/adk_web_server.py

@@ -17,6 +17,7 @@ from __future__ import annotations
 import asyncio
 from contextlib import asynccontextmanager
 import importlib
+import ipaddress
 import json
 import logging
 import os
@@ -29,7 +30,9 @@ from typing import Any
 from typing import Callable
 from typing import List
 from typing import Literal
+from typing import Mapping
 from typing import Optional
+import urllib.parse
 
 from fastapi import FastAPI
 from fastapi import HTTPException
@@ -190,6 +193,47 @@ def _get_scope_header(
   return None
 
 
+_LOOPBACK_HOSTNAMES = frozenset({"localhost"})
+
+
+def _strip_port(host: str) -> str:
+  """Returns *host* without its port, or unchanged if it has no valid one."""
+  # A malformed authority must come back whole, so that callers never read
+  # "127.0.0.1:8000.evil.com" as loopback.
+  if host.startswith("["):  # [addr] or [addr]:port
+    bare, bracket, suffix = host[1:].partition("]")
+    if not bracket:
+      return host
+  elif host.count(":") == 1:  # host:port; bracketless IPv6 has more colons
+    bare, _, port = host.partition(":")
+    suffix = f":{port}"
+  else:
+    return host
+  if suffix and not (suffix.startswith(":") and suffix[1:].isdigit()):
+    return host
+  return bare
+
+
+def _is_loopback_address(host: str) -> bool:
+  """Return True if *host* (with or without a port) refers to a loopback address."""
+  # Host names are case-insensitive and may carry a root dot ("localhost.").
+  bare = _strip_port(host).lower().rstrip(".")
+  if bare in _LOOPBACK_HOSTNAMES:
+    return True
+  try:
+    return ipaddress.ip_address(bare).is_loopback
+  except ValueError:
+    return False
+
+
+def _get_server_host(scope: dict[str, Any]) -> Optional[str]:
+  """Return the host the server is actually bound to (from ASGI server port)."""
+  server = scope.get("server")
+  if server and len(server) == 2:
+    return str(server[0])
+  return None
+
+
 def _get_request_origin(scope: dict[str, Any]) -> Optional[str]:
   """Compute the effective origin for the current HTTP/WebSocket request."""
   forwarded = _get_scope_header(scope, b"forwarded")
@@ -219,30 +263,132 @@ def _get_request_origin(scope: dict[str, Any]) -> Optional[str]:
   return f"{_normalize_origin_scheme(proto)}://{host}"
 
 
+def _get_allowed_request_hosts(
+    allowed_literal_origins: list[str],
+) -> Optional[frozenset[str]]:
+  """Returns hosts the rebinding guard accepts besides loopback, None for all."""
+  # A loopback bind behind a same-machine proxy sees the proxy's hostname in
+  # Host, so listing an origin in --allow_origins vouches for its host. A
+  # 'regex:' entry yields no host, so only "*" opts out of the guard.
+  if "*" in allowed_literal_origins:
+    return None
+
+  hosts = set()
+  for origin in allowed_literal_origins:
+    try:
+      host = urllib.parse.urlparse(origin).hostname
+    except ValueError:
+      continue  # A malformed origin vouches for no host.
+    if host:
+      hosts.add(host.lower())
+  return frozenset(hosts)
+
+
+def _is_dns_rebinding_request(
+    scope: Mapping[str, Any],
+    bind_host: Optional[str],
+    allowed_request_hosts: Optional[frozenset[str]],
+) -> bool:
+  """Returns True if the request must be rejected as possible DNS rebinding."""
+  # A loopback bind is reachable only from this machine, so a request naming
+  # any other host was pointed here by rebound DNS. Origin cannot catch that:
+  # browsers omit it on requests they consider same-origin, as a rebound page's
+  # are, so callers must apply this to every request, safe methods included.
+  if allowed_request_hosts is None or bind_host is None:
+    # A bind we were not told about is not ours to guess: an app embedded
+    # behind a same-machine proxy would then reject its own traffic.
+    return False
+  if not _is_loopback_address(bind_host):
+    return False
+
+  # Only the real Host header will do: it is a forbidden request header,
+  # whereas a same-origin fetch() may set X-Forwarded-Host or Forwarded freely.
+  host_values = [
+      value.decode("latin-1").strip()
+      for name, value in scope.get("headers", [])
+      if name.lower() == b"host"
+  ]
+  if not host_values:
+    # Browsers always send Host, so its absence is not a rebinding vector.
+    return False
+  if len(host_values) > 1 or "," in host_values[0]:
+    # Host is a singleton header; a list of them is smuggling, not a client.
+    return True
+  if _is_loopback_address(host_values[0]):
+    return False
+
+  return (
+      _strip_port(host_values[0]).lower().rstrip(".")
+      not in allowed_request_hosts
+  )
+
+
 def _is_request_origin_allowed(
     origin: str,
     scope: dict[str, Any],
     allowed_literal_origins: list[str],
     allowed_origin_regex: Optional[re.Pattern[str]],
     has_configured_allowed_origins: bool,
+    bind_host: Optional[str] = None,
 ) -> bool:
-  """Validate an Origin header against explicit config or same-origin."""
+  """Validate an Origin header against explicit config or same-origin.
+
+  DNS-rebinding protection: when the server is bound to a loopback address
+  (127.0.0.1 / ::1 / localhost) and no explicit allow-origins have been
+  configured, we additionally require that the request's Origin header also
+  resolves to a loopback host.  This prevents a DNS-rebinding attack where
+  an external page temporarily resolves to 127.0.0.1 and then reaches the
+  local development server by matching its own (evil.com) origin against the
+  Host header it controls.
+  """
   if has_configured_allowed_origins and _is_origin_allowed(
       origin, allowed_literal_origins, allowed_origin_regex
   ):
     return True
 
+  # DNS-rebinding guard: if the server is on loopback and no explicit
+  # allow-origins list is configured, only permit origins whose host is also
+  # loopback.  This mirrors the protection used by the MCP go-sdk SSEHandler.
+  # scope["server"] is only a fallback for an unknown bind: ASGI servers fill
+  # it from the accepted socket, so a wildcard bind reports 127.0.0.1 here.
+  server_host = _get_server_host(scope) if bind_host is None else bind_host
+  if (
+      not has_configured_allowed_origins
+      and server_host is not None
+      and _is_loopback_address(server_host)
+  ):
+    try:
+      origin_host = urllib.parse.urlparse(origin).hostname or ""
+    except Exception:  # pylint: disable=broad-except
+      return False
+    if not _is_loopback_address(origin_host):
+      return False
+
   request_origin = _get_request_origin(scope)
   if request_origin is None:
     return False
   return origin == request_origin
 
 
-_SAFE_HTTP_METHODS = frozenset({"GET", "HEAD", "OPTIONS"})
+async def _send_forbidden(send: Any, reason: str) -> None:
+  """Sends a plain-text 403 over the ASGI send channel."""
+  response_body = f"Forbidden: {reason}".encode()
+  await send({
+      "type": "http.response.start",
+      "status": 403,
+      "headers": [
+          (b"content-type", b"text/plain"),
+          (b"content-length", str(len(response_body)).encode()),
+      ],
+  })
+  await send({
+      "type": "http.response.body",
+      "body": response_body,
+  })
 
 
 class _OriginCheckMiddleware:
-  """ASGI middleware that blocks cross-origin state-changing requests."""
+  """ASGI middleware that blocks cross-origin requests."""
 
   def __init__(
       self,
@@ -250,11 +396,14 @@ class _OriginCheckMiddleware:
       has_configured_allowed_origins: bool,
       allowed_origins: list[str],
       allowed_origin_regex: Optional[re.Pattern[str]],
+      bind_host: Optional[str] = None,
   ) -> None:
     self._app = app
     self._has_configured_allowed_origins = has_configured_allowed_origins
     self._allowed_origins = allowed_origins
     self._allowed_origin_regex = allowed_origin_regex
+    self._bind_host = bind_host
+    self._allowed_request_hosts = _get_allowed_request_hosts(allowed_origins)
 
   async def __call__(
       self,
@@ -266,39 +415,27 @@ class _OriginCheckMiddleware:
       await self._app(scope, receive, send)
       return
 
-    method = scope.get("method", "GET")
-    if method in _SAFE_HTTP_METHODS:
-      await self._app(scope, receive, send)
+    # Every method: the reads here are the whole session history, and a rebound
+    # page looks same-origin, so neither method nor Origin can gate them.
+    if _is_dns_rebinding_request(
+        scope, self._bind_host, self._allowed_request_hosts
+    ):
+      await _send_forbidden(send, "host not allowed")
       return
 
     origin = _get_scope_header(scope, b"origin")
-    if origin is None:
-      await self._app(scope, receive, send)
-      return
-
-    if _is_request_origin_allowed(
+    if origin is not None and not _is_request_origin_allowed(
         origin,
         scope,
         self._allowed_origins,
         self._allowed_origin_regex,
         self._has_configured_allowed_origins,
+        self._bind_host,
     ):
-      await self._app(scope, receive, send)
+      await _send_forbidden(send, "origin not allowed")
       return
 
-    response_body = b"Forbidden: origin not allowed"
-    await send({
-        "type": "http.response.start",
-        "status": 403,
-        "headers": [
-            (b"content-type", b"text/plain"),
-            (b"content-length", str(len(response_body)).encode()),
-        ],
-    })
-    await send({
-        "type": "http.response.body",
-        "body": response_body,
-    })
+    await self._app(scope, receive, send)
 
 
 class ApiServerSpanExporter(export_lib.SpanExporter):
@@ -938,6 +1075,7 @@ class AdkWebServer:
       register_processors: Callable[[TracerProvider], None] = lambda o: None,
       otel_to_cloud: bool = False,
       with_ui: bool = False,
+      bind_host: Optional[str] = None,
   ):
     """Creates a FastAPI app for the ADK web server.
 
@@ -959,6 +1097,9 @@ class AdkWebServer:
         to the TracerProvider.
       otel_to_cloud: Whether to enable Cloud Trace and Cloud Logging
         integrations.
+      bind_host: The address the server will bind. A loopback value rejects
+        requests addressed to any other host as DNS rebinding; None disables
+        that, for callers that do not own the bind.
 
     Returns:
       A FastAPI app instance.
@@ -1026,7 +1167,9 @@ class AdkWebServer:
         has_configured_allowed_origins=has_configured_allowed_origins,
         allowed_origins=literal_origins,
         allowed_origin_regex=compiled_origin_regex,
+        bind_host=bind_host,
     )
+    allowed_request_hosts = _get_allowed_request_hosts(literal_origins)
 
     @app.get("/health")
     async def health() -> dict[str, str]:
@@ -2130,6 +2273,13 @@ class AdkWebServer:
         enable_session_resumption: bool | None = Query(default=None),
         save_live_blob: bool = Query(default=False),
     ) -> None:
+      # Before anything else: this decides whether the caller may talk to us.
+      if _is_dns_rebinding_request(
+          websocket.scope, bind_host, allowed_request_hosts
+      ):
+        await websocket.close(code=1008, reason="Host not allowed")
+        return
+
       ws_origin = websocket.headers.get("origin")
       if ws_origin is not None and not _is_request_origin_allowed(
           ws_origin,
@@ -2137,6 +2287,7 @@ class AdkWebServer:
           literal_origins,
           compiled_origin_regex,
           has_configured_allowed_origins,
+          bind_host,
       ):
         await websocket.close(code=1008, reason="Origin not allowed")
         return

src/google/adk/cli/cli_tools_click.py

@@ -1637,6 +1637,7 @@ def cli_web(
       lifespan=_lifespan,
       a2a=a2a,
       host=host,
+      bind_host=host,
       port=port,
       url_prefix=url_prefix,
       reload_agents=reload_agents,
@@ -1723,6 +1724,7 @@ def cli_api_server(
           otel_to_cloud=otel_to_cloud,
           a2a=a2a,
           host=host,
+          bind_host=host,
           port=port,
           url_prefix=url_prefix,
           reload_agents=reload_agents,

src/google/adk/cli/fast_api.py

@@ -88,6 +88,7 @@ def get_fast_api_app(
     a2a: bool = False,
     task_store_uri: str | None = None,
     host: str = "127.0.0.1",
+    bind_host: str | None = None,
     port: int = 8000,
     url_prefix: str | None = None,
     trace_to_cloud: bool = False,
@@ -132,7 +133,11 @@ def get_fast_api_app(
     a2a: Whether to enable Agent-to-Agent (A2A) protocol support.
     task_store_uri: URI for the A2A task store. Uses in-memory task store if
       None. Only used when ``a2a=True``.
-    host: Host address for the server (defaults to 127.0.0.1).
+    host: Host address for the server (defaults to 127.0.0.1). Unused by the
+      returned app; pass ``bind_host`` to guard it.
+    bind_host: The address the caller will bind the returned app to. A loopback
+      value turns on DNS-rebinding protection, which rejects requests addressed
+      to any other host. Leave it None to serve the app yourself without that.
     port: Port number for the server (defaults to 8000).
     url_prefix: Optional prefix for all URL routes.
     trace_to_cloud: Whether to export traces to Google Cloud Trace.
@@ -307,6 +312,7 @@ def get_fast_api_app(
       lifespan=lifespan,
       allow_origins=allow_origins,
       otel_to_cloud=otel_to_cloud,
+      bind_host=bind_host,
       **extra_fast_api_args,
   )
 

tests/unittests/cli/test_adk_web_server_run_live.py

@@ -221,8 +221,11 @@ _WS_BASE_URL = (
 )
 
 
-def _build_ws_client():
-  """Build a TestClient wired to a capturing runner."""
+def _build_ws_client(bind_host=None):
+  """Build a TestClient wired to a capturing runner.
+
+  A loopback *bind_host* turns on the DNS-rebinding guard.
+  """
   session_service = InMemorySessionService()
   asyncio.run(
       session_service.create_session(
@@ -253,10 +256,26 @@ def _build_ws_client():
   fast_api_app = adk_web_server.get_fast_api_app(
       setup_observer=lambda _observer, _server: None,
       tear_down_observer=lambda _observer, _server: None,
+      bind_host=bind_host,
   )
   return TestClient(fast_api_app)
 
 
+def test_run_live_rejects_rebound_host():
+  """A rebound handshake carries the attacker's hostname in Host."""
+  client = _build_ws_client(bind_host="127.0.0.1")
+  with pytest.raises(WebSocketDisconnect) as exc_info:
+    with client.websocket_connect(f"ws://evil.com:8000{_WS_BASE_URL}") as ws:
+      ws.receive_text()
+  assert exc_info.value.code == 1008
+
+
+def test_run_live_allows_loopback_host():
+  client = _build_ws_client(bind_host="127.0.0.1")
+  with client.websocket_connect(f"ws://localhost:8000{_WS_BASE_URL}") as ws:
+    _ = ws.receive_text()
+
+
 def test_run_live_rejects_disallowed_origin():
   client = _build_ws_client()
   with pytest.raises(WebSocketDisconnect) as exc_info:

tests/unittests/cli/test_dns_rebinding_protection.py

@@ -0,0 +1,481 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Tests for DNS-rebinding protection in _OriginCheckMiddleware."""
+
+import asyncio
+from typing import Any
+from typing import Optional
+
+from google.adk.cli.adk_web_server import _get_allowed_request_hosts
+from google.adk.cli.adk_web_server import _is_dns_rebinding_request
+from google.adk.cli.adk_web_server import _is_loopback_address
+from google.adk.cli.adk_web_server import _is_request_origin_allowed
+from google.adk.cli.adk_web_server import _OriginCheckMiddleware
+import pytest
+
+
+class TestIsLoopbackAddress:
+  """Unit tests for _is_loopback_address."""
+
+  @pytest.mark.parametrize(
+      "host",
+      [
+          "127.0.0.1",
+          "localhost",
+          "::1",
+          "[::1]",
+          "127.0.0.1:8000",
+          "localhost:8000",
+          "[::1]:8000",
+          "127.1.2.3",  # any 127.x.x.x is loopback
+      ],
+  )
+  def test_loopback_hosts(self, host: str):
+    assert _is_loopback_address(host), f"{host!r} should be loopback"
+
+  @pytest.mark.parametrize(
+      "host",
+      [
+          "evil.com",
+          "127.evil.com",
+          "0.0.0.0",
+          "192.168.1.1",
+          "10.0.0.1",
+          "128.0.0.1",
+          "",
+      ],
+  )
+  def test_non_loopback_hosts(self, host: str):
+    assert not _is_loopback_address(host), f"{host!r} should NOT be loopback"
+
+
+class TestDnsRebindingProtection:
+  """Tests that DNS-rebinding attacks are blocked when server is on loopback."""
+
+  def _make_scope(
+      self, server_host: str = "127.0.0.1", host_header: str = "127.0.0.1:8000"
+  ) -> dict:
+    """Build a minimal ASGI scope for testing."""
+    return {
+        "type": "http",
+        "method": "POST",
+        "server": (server_host, 8000),
+        "headers": [
+            (b"host", host_header.encode()),
+        ],
+        "scheme": "http",
+    }
+
+  # --- DNS rebinding scenarios (should be BLOCKED) ---
+
+  def test_dns_rebinding_evil_origin_loopback_server_no_configured_origins(
+      self,
+  ):
+    """Attacker page (evil.com) DNS-rebinds to 127.0.0.1 and sends a POST.
+
+    Browser sends Origin: http://evil.com, Host: evil.com.
+    Server is bound to 127.0.0.1.
+    No explicit allow-origins configured.
+    Expected: BLOCKED.
+    """
+    scope = self._make_scope(
+        server_host="127.0.0.1", host_header="evil.com:8000"
+    )
+    result = _is_request_origin_allowed(
+        origin="http://evil.com",
+        scope=scope,
+        allowed_literal_origins=[],
+        allowed_origin_regex=None,
+        has_configured_allowed_origins=False,
+    )
+    assert (
+        not result
+    ), "DNS-rebinding from evil.com should be blocked on loopback server"
+
+  def test_dns_rebinding_127_evil_origin(self):
+    """Origin header host starts with '127.' but is a hostname (127.evil.com)."""
+    scope = self._make_scope(
+        server_host="127.0.0.1", host_header="127.evil.com:8000"
+    )
+    result = _is_request_origin_allowed(
+        origin="http://127.evil.com",
+        scope=scope,
+        allowed_literal_origins=[],
+        allowed_origin_regex=None,
+        has_configured_allowed_origins=False,
+    )
+    assert not result
+
+  def test_dns_rebinding_localhost_server(self):
+    """Same attack, server bound as 'localhost'."""
+    scope = self._make_scope(server_host="localhost", host_header="evil.com")
+    result = _is_request_origin_allowed(
+        origin="http://evil.com",
+        scope=scope,
+        allowed_literal_origins=[],
+        allowed_origin_regex=None,
+        has_configured_allowed_origins=False,
+    )
+    assert not result
+
+  def test_dns_rebinding_ipv6_loopback_server(self):
+    """Same attack, server bound to ::1."""
+    scope = self._make_scope(server_host="::1", host_header="evil.com")
+    result = _is_request_origin_allowed(
+        origin="http://evil.com",
+        scope=scope,
+        allowed_literal_origins=[],
+        allowed_origin_regex=None,
+        has_configured_allowed_origins=False,
+    )
+    assert not result
+
+  # --- Legitimate same-origin requests (should be ALLOWED) ---
+
+  def test_same_origin_localhost_allowed(self):
+    """Legitimate browser request from localhost UI to localhost server."""
+    scope = self._make_scope(
+        server_host="127.0.0.1", host_header="127.0.0.1:8000"
+    )
+    result = _is_request_origin_allowed(
+        origin="http://127.0.0.1:8000",
+        scope=scope,
+        allowed_literal_origins=[],
+        allowed_origin_regex=None,
+        has_configured_allowed_origins=False,
+    )
+    assert result, "Same-origin localhost request should be allowed"
+
+  def test_same_origin_localhost_named(self):
+    """Browser opens http://localhost:8000 -> requests to localhost:8000."""
+    scope = self._make_scope(
+        server_host="127.0.0.1", host_header="localhost:8000"
+    )
+    result = _is_request_origin_allowed(
+        origin="http://localhost:8000",
+        scope=scope,
+        allowed_literal_origins=[],
+        allowed_origin_regex=None,
+        has_configured_allowed_origins=False,
+    )
+    assert result
+
+  # --- Explicit allow-origins configured (allow-list bypasses DNS guard) ---
+
+  def test_explicit_allowlist_overrides_dns_rebinding_guard(self):
+    """If the developer explicitly allows evil.com, it should be permitted."""
+    scope = self._make_scope(server_host="127.0.0.1", host_header="evil.com")
+    result = _is_request_origin_allowed(
+        origin="http://evil.com",
+        scope=scope,
+        allowed_literal_origins=["http://evil.com"],
+        allowed_origin_regex=None,
+        has_configured_allowed_origins=True,
+    )
+    assert result, "Explicitly allowed origin should still pass"
+
+  # --- Non-loopback server (protection does not apply) ---
+
+  def test_non_loopback_server_no_dns_guard(self):
+    """Server bound to 0.0.0.0 — DNS guard must not interfere with same-origin check."""
+    scope = self._make_scope(
+        server_host="0.0.0.0", host_header="example.com:8000"
+    )
+    result = _is_request_origin_allowed(
+        origin="http://example.com:8000",
+        scope=scope,
+        allowed_literal_origins=[],
+        allowed_origin_regex=None,
+        has_configured_allowed_origins=False,
+    )
+    assert result, "Same-origin on public server should be allowed"
+
+
+def _make_http_scope(
+    method: str = "GET",
+    server_host: str = "127.0.0.1",
+    host_header: Optional[str] = "127.0.0.1:8000",
+    origin: Optional[str] = None,
+    extra_headers: Optional[list[tuple[bytes, bytes]]] = None,
+) -> dict[str, Any]:
+  """Builds a minimal ASGI HTTP scope."""
+  # server_host is the local end of the connection, which is what ASGI servers
+  # report, not the address the server was told to bind.
+  headers: list[tuple[bytes, bytes]] = []
+  if host_header is not None:
+    headers.append((b"host", host_header.encode()))
+  if origin is not None:
+    headers.append((b"origin", origin.encode()))
+  headers.extend(extra_headers or [])
+  return {
+      "type": "http",
+      "method": method,
+      "server": (server_host, 8000),
+      "headers": headers,
+      "scheme": "http",
+  }
+
+
+class TestGetAllowedRequestHosts:
+  """Unit tests for deriving accepted Host values from --allow_origins."""
+
+  def test_no_configuration_accepts_nothing_extra(self):
+    assert _get_allowed_request_hosts([]) == frozenset()
+
+  def test_literal_origins_contribute_their_hosts(self):
+    """Hosts are compared case-insensitively, so they are folded here."""
+    assert _get_allowed_request_hosts(
+        ["https://Proxy.Example.COM", "http://localhost:3000"]
+    ) == frozenset({"proxy.example.com", "localhost"})
+
+  def test_entry_without_a_host_contributes_nothing(self):
+    """A scheme-less or unparsable entry has no hostname to vouch for."""
+    assert (
+        _get_allowed_request_hosts(["localhost:3000", "", "http://[::1"])
+        == frozenset()
+    )
+
+  def test_only_wildcard_disables_the_guard(self):
+    """A wildcard already says "accept anything" out loud."""
+    assert _get_allowed_request_hosts(["*"]) is None
+
+
+class TestIsDnsRebindingRequest:
+  """Unit tests for the Host-header based DNS-rebinding guard."""
+
+  def test_rebound_host_on_loopback_bind_is_rejected(self):
+    """The attacker's domain in Host, while we are bound to loopback."""
+    scope = _make_http_scope(host_header="evil.com:8000")
+    assert _is_dns_rebinding_request(scope, "127.0.0.1", frozenset())
+
+  @pytest.mark.parametrize(
+      "host_header", ["localhost:8000", "127.0.0.1:8000", "[::1]:8000"]
+  )
+  def test_loopback_host_is_accepted(self, host_header):
+    scope = _make_http_scope(host_header=host_header)
+    assert not _is_dns_rebinding_request(scope, "127.0.0.1", frozenset())
+
+  def test_forwarded_headers_cannot_vouch_for_the_host(self):
+    """Regression: a rebound page can set these, so only Host can be trusted."""
+    for spoofed in [
+        (b"x-forwarded-host", b"127.0.0.1:8000"),
+        (b"forwarded", b"proto=http;host=127.0.0.1:8000"),
+        (b"x-forwarded-host", b"127.0.0.1, evil.com"),
+    ]:
+      scope = _make_http_scope(
+          host_header="evil.com:8000", extra_headers=[spoofed]
+      )
+      assert _is_dns_rebinding_request(
+          scope, "127.0.0.1", frozenset()
+      ), f"{spoofed!r} must not overrule the Host header"
+
+  def test_host_from_allow_origins_is_accepted(self):
+    """A same-machine reverse proxy is named via --allow_origins."""
+    scope = _make_http_scope(host_header="proxy.example.com")
+    assert not _is_dns_rebinding_request(
+        scope, "127.0.0.1", frozenset({"proxy.example.com"})
+    )
+
+  def test_allow_origins_does_not_vouch_for_other_hosts(self):
+    """Regression: configuring an origin must not disable the guard wholesale."""
+    scope = _make_http_scope(host_header="evil.com:8000")
+    assert _is_dns_rebinding_request(
+        scope, "127.0.0.1", frozenset({"proxy.example.com"})
+    )
+
+  def test_blanket_allow_origins_disables_the_guard(self):
+    scope = _make_http_scope(host_header="evil.com:8000")
+    assert not _is_dns_rebinding_request(scope, "127.0.0.1", None)
+
+  @pytest.mark.parametrize("bind_host", ["0.0.0.0", "::", "192.168.1.5"])
+  def test_non_loopback_bind_is_not_guarded(self, bind_host):
+    """`adk deploy` binds a public interface to serve other hosts on purpose."""
+    scope = _make_http_scope(host_header="my-service.run.app")
+    assert not _is_dns_rebinding_request(scope, bind_host, frozenset())
+
+  def test_wildcard_bind_reached_over_loopback_is_not_guarded(self):
+    """Regression: scope["server"] is the accepted socket, not the bind.
+
+    A wildcard bind reports 127.0.0.1 for a loopback connection, which is what
+    a same-host proxy makes; keying off it would 403 every one.
+    """
+    scope = _make_http_scope(
+        server_host="127.0.0.1", host_header="my-service.run.app"
+    )
+    assert not _is_dns_rebinding_request(scope, "0.0.0.0", frozenset())
+
+  def test_unknown_bind_is_not_guarded(self):
+    """Regression: guessing an embedded app's bind would 403 its own traffic."""
+    scope = _make_http_scope(
+        server_host="127.0.0.1", host_header="evil.com:8000"
+    )
+    assert not _is_dns_rebinding_request(scope, None, frozenset())
+
+  @pytest.mark.parametrize(
+      "host_header",
+      [
+          "127.0.0.1, evil.com",
+          "evil.com, 127.0.0.1",
+          "[::1].evil.com",
+          "[::1]evil.com",
+          "[::1",
+          "127.0.0.1:8000.evil.com",
+          "localhost:8000x",
+          "[127.0.0.1]@evil.com",
+      ],
+  )
+  def test_smuggled_host_is_rejected(self, host_header):
+    """A single loopback-looking token must not vouch for the whole header."""
+    scope = _make_http_scope(host_header=host_header)
+    assert _is_dns_rebinding_request(scope, "127.0.0.1", frozenset())
+
+  def test_duplicate_host_headers_are_rejected(self):
+    """The loopback one comes first, so only the singleton rule can reject."""
+    scope = _make_http_scope(
+        host_header="127.0.0.1:8000",
+        extra_headers=[(b"host", b"evil.com:8000")],
+    )
+    assert _is_dns_rebinding_request(scope, "127.0.0.1", frozenset())
+
+  @pytest.mark.parametrize(
+      "host_header", ["LOCALHOST:8000", "localhost.:8000", "LocalHost."]
+  )
+  def test_loopback_host_spellings_are_accepted(self, host_header):
+    """Host names are case-insensitive and may carry the root dot."""
+    scope = _make_http_scope(host_header=host_header)
+    assert not _is_dns_rebinding_request(scope, "127.0.0.1", frozenset())
+
+  def test_missing_host_header_is_accepted(self):
+    """Non-browser clients may omit Host; they are not a rebinding vector."""
+    scope = _make_http_scope(host_header=None)
+    assert not _is_dns_rebinding_request(scope, "127.0.0.1", frozenset())
+
+
+class TestOriginCheckMiddleware:
+  """End-to-end checks that reads are guarded, not just state-changing calls."""
+
+  def _call(
+      self,
+      scope: dict[str, Any],
+      bind_host: Optional[str] = "127.0.0.1",
+      allow_origins: Optional[list[str]] = None,
+  ) -> tuple[Optional[int], bool]:
+    """Returns (status code, whether the wrapped app was reached)."""
+    reached = False
+    statuses: list[int] = []
+
+    async def inner_app(scope, receive, send):
+      del receive
+      nonlocal reached
+      reached = True
+      if scope["type"] == "http":
+        await send({"type": "http.response.start", "status": 200})
+        await send({"type": "http.response.body", "body": b"ok"})
+
+    async def send(message):
+      if message["type"] == "http.response.start":
+        statuses.append(message["status"])
+
+    async def receive():
+      return {"type": "http.request", "body": b"", "more_body": False}
+
+    middleware = _OriginCheckMiddleware(
+        inner_app,
+        has_configured_allowed_origins=bool(allow_origins),
+        allowed_origins=allow_origins or [],
+        allowed_origin_regex=None,
+        bind_host=bind_host,
+    )
+    asyncio.run(middleware(scope, receive, send))
+    return (statuses[0] if statuses else None), reached
+
+  def test_proxy_host_named_in_allow_origins_is_served(self):
+    """A loopback bind behind a same-machine proxy names the proxy origin."""
+    status, reached = self._call(
+        _make_http_scope(
+            host_header="proxy.example.com",
+            origin="https://proxy.example.com",
+        ),
+        allow_origins=["https://proxy.example.com"],
+    )
+    assert status == 200
+    assert reached
+
+  @pytest.mark.parametrize("origin", [None, "http://evil.com:8000"])
+  @pytest.mark.parametrize("method", ["GET", "HEAD", "OPTIONS", "POST"])
+  def test_rebound_host_is_blocked_for_every_method(self, method, origin):
+    """Regression: reads, and requests without Origin, both skipped the check."""
+    status, reached = self._call(
+        _make_http_scope(
+            method=method, host_header="evil.com:8000", origin=origin
+        )
+    )
+    assert status == 403
+    assert not reached
+
+  @pytest.mark.parametrize("method", ["GET", "HEAD", "OPTIONS", "POST"])
+  def test_same_origin_dev_ui_still_allowed(self, method: str):
+    status, reached = self._call(
+        _make_http_scope(
+            method=method,
+            host_header="localhost:8000",
+            origin="http://localhost:8000",
+        )
+    )
+    assert status == 200
+    assert reached
+
+  def test_local_request_without_origin_allowed(self):
+    """curl, the ADK CLI and same-origin browser reads send no Origin."""
+    status, reached = self._call(_make_http_scope(host_header="127.0.0.1:8000"))
+    assert status == 200
+    assert reached
+
+  def test_cross_origin_get_with_foreign_origin_is_blocked(self):
+    """A read carrying a foreign Origin is no longer waved through."""
+    status, reached = self._call(
+        _make_http_scope(host_header="127.0.0.1:8000", origin="http://evil.com")
+    )
+    assert status == 403
+    assert not reached
+
+  def test_configured_origin_allowed_for_reads(self):
+    status, reached = self._call(
+        _make_http_scope(
+            host_header="127.0.0.1:8000", origin="http://localhost:3000"
+        ),
+        allow_origins=["http://localhost:3000"],
+    )
+    assert status == 200
+    assert reached
+
+  def test_public_bind_same_origin_still_allowed(self):
+    """`adk deploy` containers bind 0.0.0.0 and serve a real hostname."""
+    status, reached = self._call(
+        _make_http_scope(
+            host_header="my-service.run.app",
+            origin="http://my-service.run.app",
+        ),
+        bind_host="0.0.0.0",
+    )
+    assert status == 200
+    assert reached
+
+  def test_non_http_scope_is_passed_through(self):
+    """Lifespan messages are not requests and carry nothing to validate."""
+    scope = _make_http_scope(host_header="evil.com:8000")
+    scope["type"] = "lifespan"
+    _, reached = self._call(scope)
+    assert reached

tests/unittests/cli/test_fast_api.py

@@ -57,6 +57,14 @@ logging.basicConfig(
 )
 logger = logging.getLogger("google_adk." + __name__)
 
+# An app told it binds 127.0.0.1 rejects requests addressed to any other host,
+# so its client cannot use TestClient's default "http://testserver".
+_LOOPBACK_BASE_URL = "http://127.0.0.1:8000"
+
+# What a browser addresses when a hosted dev environment forwards its port to a
+# loopback bind.
+_PROXY_ORIGIN = "https://8000-my-workstation.example.dev"
+
 
 # Here we create a dummy agent module that get_fast_api_app expects
 class DummyAgent(BaseAgent):
@@ -724,9 +732,10 @@ def builder_test_client(
         allow_origins=None,
         a2a=False,
         host="127.0.0.1",
+        bind_host="127.0.0.1",
         port=8000,
     )
-    return TestClient(app)
+    return TestClient(app, base_url=_LOOPBACK_BASE_URL)
 
 
 @pytest.fixture
@@ -2358,7 +2367,7 @@ def test_builder_save_rejects_cross_origin_post(builder_test_client, tmp_path):
 def test_builder_save_allows_same_origin_post(builder_test_client, tmp_path):
   response = builder_test_client.post(
       "/builder/save?tmp=true",
-      headers={"origin": "http://testserver"},
+      headers={"origin": _LOOPBACK_BASE_URL},
       files=[(
           "files",
           ("app/root_agent.yaml", b"name: app\n", "application/x-yaml"),
@@ -2370,14 +2379,91 @@ def test_builder_save_allows_same_origin_post(builder_test_client, tmp_path):
   assert (tmp_path / "app" / "tmp" / "app" / "root_agent.yaml").is_file()
 
 
-def test_builder_get_allows_cross_origin_get(builder_test_client):
+def test_builder_get_rejects_cross_origin_get(builder_test_client):
+  """Reads expose agent config and session data, so they are guarded too."""
   response = builder_test_client.get(
       "/builder/app/missing?tmp=true",
       headers={"origin": "https://evil.com"},
   )
 
+  assert response.status_code == 403
+  assert response.text == "Forbidden: origin not allowed"
+
+
+def test_builder_get_allows_same_origin_get(builder_test_client):
+  """The dev UI reads its own agent config from the same origin."""
+  response = builder_test_client.get(
+      "/builder/app/missing?tmp=true",
+      headers={"origin": _LOOPBACK_BASE_URL},
+  )
+
   assert response.status_code == 200
-  assert response.text == ""
+  assert not response.text
+
+
+def test_builder_get_allows_request_without_origin(builder_test_client):
+  """Browsers omit Origin on same-origin reads, and CLI clients never send it."""
+  response = builder_test_client.get("/builder/app/missing?tmp=true")
+
+  assert response.status_code == 200
+  assert not response.text
+
+
+def test_proxied_host_named_in_allow_origins_is_served(
+    mock_session_service,
+    mock_artifact_service,
+    mock_memory_service,
+    mock_agent_loader,
+    mock_eval_sets_manager,
+    mock_eval_set_results_manager,
+):
+  """A hosted dev environment forwards the browser's own hostname in Host.
+
+  The server still binds loopback there, so the page it serves is reachable
+  only by naming that hostname in allow_origins.
+  """
+  client = _create_test_client(
+      mock_session_service,
+      mock_artifact_service,
+      mock_memory_service,
+      mock_agent_loader,
+      mock_eval_sets_manager,
+      mock_eval_set_results_manager,
+      allow_origins=[_PROXY_ORIGIN],
+      bind_host="127.0.0.1",
+  )
+
+  assert client.get(f"{_PROXY_ORIGIN}/health").status_code == 200
+  # The index page is behind the same middleware, so it would 403 too, and the
+  # dev UI would not load at all.
+  index = client.get(f"{_PROXY_ORIGIN}/", follow_redirects=False)
+  assert index.status_code == 307
+
+
+def test_proxied_host_absent_from_allow_origins_is_rejected(
+    mock_session_service,
+    mock_artifact_service,
+    mock_memory_service,
+    mock_agent_loader,
+    mock_eval_sets_manager,
+    mock_eval_set_results_manager,
+):
+  """Nothing distinguishes an unnamed proxy hostname from a rebound one."""
+  client = _create_test_client(
+      mock_session_service,
+      mock_artifact_service,
+      mock_memory_service,
+      mock_agent_loader,
+      mock_eval_sets_manager,
+      mock_eval_set_results_manager,
+      allow_origins=None,
+      bind_host="127.0.0.1",
+  )
+
+  response = client.get(f"{_PROXY_ORIGIN}/health")
+
+  assert response.status_code == 403
+  assert response.text == "Forbidden: host not allowed"
 
 
 def test_builder_cancel_deletes_tmp_idempotent(builder_test_client, tmp_path):

tests/unittests/cli/utils/test_cli_tools_click.py

@@ -646,6 +646,56 @@ def test_cli_migrate_session_allows_unsafe_unpickling_flag(
   }]
 
 
+@pytest.mark.parametrize("command", ["web", "api_server"])
+@pytest.mark.parametrize("host", ["127.0.0.1", "0.0.0.0"])
+def test_cli_arms_rebinding_guard_with_the_address_it_binds(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, command: str, host: str
+) -> None:
+  """The DNS-rebinding guard is off unless the CLI names the address it binds.
+
+  Every other test of the guard supplies ``bind_host`` itself, so only this one
+  fails if the CLI stops passing it and serves `adk web` unguarded again.
+  """
+  agents_dir = tmp_path / "agents"
+  agents_dir.mkdir()
+
+  app_kwargs: Dict[str, Any] = {}
+  uvicorn_kwargs: Dict[str, Any] = {}
+
+  def _record_get_fast_api_app(**kwargs: Any) -> object:
+    app_kwargs.update(kwargs)
+    return object()
+
+  def _record_uvicorn_config(*_a: Any, **kwargs: Any) -> object:
+    uvicorn_kwargs.update(kwargs)
+    return object()
+
+  class _DummyServer:
+
+    def __init__(self, *a: Any, **k: Any) -> None:
+      ...
+
+    def run(self) -> None:
+      ...
+
+  monkeypatch.setattr(
+      cli_tools_click, "get_fast_api_app", _record_get_fast_api_app
+  )
+  monkeypatch.setattr(cli_tools_click.uvicorn, "Config", _record_uvicorn_config)
+  monkeypatch.setattr(
+      cli_tools_click.uvicorn, "Server", lambda *_a, **_k: _DummyServer()
+  )
+
+  runner = CliRunner()
+  result = runner.invoke(
+      cli_tools_click.main, [command, "--host", host, str(agents_dir)]
+  )
+
+  assert result.exit_code == 0
+  assert uvicorn_kwargs.get("host") == host, "the CLI binds --host"
+  assert app_kwargs.get("bind_host") == host
+
+
 def test_cli_eval_with_eval_set_file_path(
     mock_load_eval_set_from_file,
     mock_get_root_agent,