Skip to content

Commit 9a4f479

Browse files
XananasX7copybara-github
authored andcommitted
fix: add DNS-rebinding protection to _OriginCheckMiddleware
Merge #6227 When the ADK web server is on loopback and no explicit allow-origins list is configured, require that the Origin header also resolves to a loopback host to prevent DNS-rebinding attacks. PiperOrigin-RevId: 943548404
1 parent 79b8923 commit 9a4f479

2 files changed

Lines changed: 266 additions & 1 deletion

File tree

src/google/adk/cli/api_server.py

Lines changed: 70 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -165,6 +165,48 @@ def _get_scope_header(
165165
return None
166166

167167

168+
import ipaddress as _ipaddress
169+
170+
_LOOPBACK_HOSTNAMES = frozenset({"localhost"})
171+
172+
173+
def _is_loopback_address(host: str) -> bool:
174+
"""Return True if *host* (with or without a port) refers to a loopback address.
175+
176+
Handles all four forms produced by browsers and uvicorn:
177+
- Plain IPv4: "127.0.0.1"
178+
- IPv4 with port: "127.0.0.1:8000"
179+
- Bracketed IPv6: "[::1]"
180+
- Bracketed IPv6+port: "[::1]:8000"
181+
- Plain IPv6 (scope): "::1" (ASGI server tuple value)
182+
- Hostname: "localhost"
183+
- Hostname with port: "localhost:8000"
184+
"""
185+
bare = host
186+
if bare.startswith("["):
187+
# Bracketed IPv6: [addr] or [addr]:port
188+
end = bare.find("]")
189+
if end != -1:
190+
bare = bare[1:end]
191+
elif bare.count(":") == 1:
192+
# IPv4:port or hostname:port (IPv6 without brackets has > 1 colon)
193+
bare = bare.rsplit(":", 1)[0]
194+
if bare in _LOOPBACK_HOSTNAMES:
195+
return True
196+
try:
197+
return _ipaddress.ip_address(bare).is_loopback
198+
except ValueError:
199+
return False
200+
201+
202+
def _get_server_host(scope: dict[str, Any]) -> Optional[str]:
203+
"""Return the host the server is actually bound to (from ASGI server port)."""
204+
server = scope.get("server")
205+
if server and len(server) == 2:
206+
return str(server[0])
207+
return None
208+
209+
168210
def _get_request_origin(scope: dict[str, Any]) -> Optional[str]:
169211
"""Compute the effective origin for the current HTTP/WebSocket request."""
170212
forwarded = _get_scope_header(scope, b"forwarded")
@@ -201,12 +243,39 @@ def _is_request_origin_allowed(
201243
allowed_origin_regex: Optional[re.Pattern[str]],
202244
has_configured_allowed_origins: bool,
203245
) -> bool:
204-
"""Validate an Origin header against explicit config or same-origin."""
246+
"""Validate an Origin header against explicit config or same-origin.
247+
248+
DNS-rebinding protection: when the server is bound to a loopback address
249+
(127.0.0.1 / ::1 / localhost) and no explicit allow-origins have been
250+
configured, we additionally require that the request's Origin header also
251+
resolves to a loopback host. This prevents a DNS-rebinding attack where
252+
an external page temporarily resolves to 127.0.0.1 and then POSTs to the
253+
local development server by matching its own (evil.com) origin against the
254+
Host header it controls.
255+
"""
205256
if has_configured_allowed_origins and _is_origin_allowed(
206257
origin, allowed_literal_origins, allowed_origin_regex
207258
):
208259
return True
209260

261+
# DNS-rebinding guard: if the server is on loopback and no explicit
262+
# allow-origins list is configured, only permit origins whose host is also
263+
# loopback. This mirrors the protection used by the MCP go-sdk SSEHandler.
264+
server_host = _get_server_host(scope)
265+
if (
266+
not has_configured_allowed_origins
267+
and server_host is not None
268+
and _is_loopback_address(server_host)
269+
):
270+
try:
271+
from urllib.parse import urlparse # noqa: PLC0415 (local import OK here)
272+
273+
origin_host = urlparse(origin).hostname or ""
274+
except Exception: # pylint: disable=broad-except
275+
return False
276+
if not _is_loopback_address(origin_host):
277+
return False
278+
210279
request_origin = _get_request_origin(scope)
211280
if request_origin is None:
212281
return False
Lines changed: 196 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,196 @@
1+
# Copyright 2026 Google LLC
2+
#
3+
# Licensed under the Apache License, Version 2.0 (the "License");
4+
# you may not use this file except in compliance with the License.
5+
# You may obtain a copy of the License at
6+
#
7+
# http://www.apache.org/licenses/LICENSE-2.0
8+
#
9+
# Unless required by applicable law or agreed to in writing, software
10+
# distributed under the License is distributed on an "AS IS" BASIS,
11+
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
# See the License for the specific language governing permissions and
13+
# limitations under the License.
14+
15+
"""Tests for DNS-rebinding protection in _OriginCheckMiddleware."""
16+
17+
from google.adk.cli.api_server import _is_loopback_address
18+
from google.adk.cli.api_server import _is_request_origin_allowed
19+
import pytest
20+
21+
22+
class TestIsLoopbackAddress:
23+
"""Unit tests for _is_loopback_address."""
24+
25+
@pytest.mark.parametrize(
26+
"host",
27+
[
28+
"127.0.0.1",
29+
"localhost",
30+
"::1",
31+
"[::1]",
32+
"127.0.0.1:8000",
33+
"localhost:8000",
34+
"[::1]:8000",
35+
"127.1.2.3", # any 127.x.x.x is loopback
36+
],
37+
)
38+
def test_loopback_hosts(self, host: str):
39+
assert _is_loopback_address(host), f"{host!r} should be loopback"
40+
41+
@pytest.mark.parametrize(
42+
"host",
43+
[
44+
"evil.com",
45+
"127.evil.com",
46+
"0.0.0.0",
47+
"192.168.1.1",
48+
"10.0.0.1",
49+
"128.0.0.1",
50+
"",
51+
],
52+
)
53+
def test_non_loopback_hosts(self, host: str):
54+
assert not _is_loopback_address(host), f"{host!r} should NOT be loopback"
55+
56+
57+
class TestDnsRebindingProtection:
58+
"""Tests that DNS-rebinding attacks are blocked when server is on loopback."""
59+
60+
def _make_scope(
61+
self, server_host: str = "127.0.0.1", host_header: str = "127.0.0.1:8000"
62+
) -> dict:
63+
"""Build a minimal ASGI scope for testing."""
64+
return {
65+
"type": "http",
66+
"method": "POST",
67+
"server": (server_host, 8000),
68+
"headers": [
69+
(b"host", host_header.encode()),
70+
],
71+
"scheme": "http",
72+
}
73+
74+
# --- DNS rebinding scenarios (should be BLOCKED) ---
75+
76+
def test_dns_rebinding_evil_origin_loopback_server_no_configured_origins(
77+
self,
78+
):
79+
"""Attacker page (evil.com) DNS-rebinds to 127.0.0.1 and sends a POST.
80+
81+
Browser sends Origin: http://evil.com, Host: evil.com.
82+
Server is bound to 127.0.0.1.
83+
No explicit allow-origins configured.
84+
Expected: BLOCKED.
85+
"""
86+
scope = self._make_scope(
87+
server_host="127.0.0.1", host_header="evil.com:8000"
88+
)
89+
result = _is_request_origin_allowed(
90+
origin="http://evil.com",
91+
scope=scope,
92+
allowed_literal_origins=[],
93+
allowed_origin_regex=None,
94+
has_configured_allowed_origins=False,
95+
)
96+
assert (
97+
not result
98+
), "DNS-rebinding from evil.com should be blocked on loopback server"
99+
100+
def test_dns_rebinding_127_evil_origin(self):
101+
"""Origin header host starts with '127.' but is a hostname (127.evil.com)."""
102+
scope = self._make_scope(
103+
server_host="127.0.0.1", host_header="127.evil.com:8000"
104+
)
105+
result = _is_request_origin_allowed(
106+
origin="http://127.evil.com",
107+
scope=scope,
108+
allowed_literal_origins=[],
109+
allowed_origin_regex=None,
110+
has_configured_allowed_origins=False,
111+
)
112+
assert not result
113+
114+
def test_dns_rebinding_localhost_server(self):
115+
"""Same attack, server bound as 'localhost'."""
116+
scope = self._make_scope(server_host="localhost", host_header="evil.com")
117+
result = _is_request_origin_allowed(
118+
origin="http://evil.com",
119+
scope=scope,
120+
allowed_literal_origins=[],
121+
allowed_origin_regex=None,
122+
has_configured_allowed_origins=False,
123+
)
124+
assert not result
125+
126+
def test_dns_rebinding_ipv6_loopback_server(self):
127+
"""Same attack, server bound to ::1."""
128+
scope = self._make_scope(server_host="::1", host_header="evil.com")
129+
result = _is_request_origin_allowed(
130+
origin="http://evil.com",
131+
scope=scope,
132+
allowed_literal_origins=[],
133+
allowed_origin_regex=None,
134+
has_configured_allowed_origins=False,
135+
)
136+
assert not result
137+
138+
# --- Legitimate same-origin requests (should be ALLOWED) ---
139+
140+
def test_same_origin_localhost_allowed(self):
141+
"""Legitimate browser request from localhost UI to localhost server."""
142+
scope = self._make_scope(
143+
server_host="127.0.0.1", host_header="127.0.0.1:8000"
144+
)
145+
result = _is_request_origin_allowed(
146+
origin="http://127.0.0.1:8000",
147+
scope=scope,
148+
allowed_literal_origins=[],
149+
allowed_origin_regex=None,
150+
has_configured_allowed_origins=False,
151+
)
152+
assert result, "Same-origin localhost request should be allowed"
153+
154+
def test_same_origin_localhost_named(self):
155+
"""Browser opens http://localhost:8000 -> requests to localhost:8000."""
156+
scope = self._make_scope(
157+
server_host="127.0.0.1", host_header="localhost:8000"
158+
)
159+
result = _is_request_origin_allowed(
160+
origin="http://localhost:8000",
161+
scope=scope,
162+
allowed_literal_origins=[],
163+
allowed_origin_regex=None,
164+
has_configured_allowed_origins=False,
165+
)
166+
assert result
167+
168+
# --- Explicit allow-origins configured (allow-list bypasses DNS guard) ---
169+
170+
def test_explicit_allowlist_overrides_dns_rebinding_guard(self):
171+
"""If the developer explicitly allows evil.com, it should be permitted."""
172+
scope = self._make_scope(server_host="127.0.0.1", host_header="evil.com")
173+
result = _is_request_origin_allowed(
174+
origin="http://evil.com",
175+
scope=scope,
176+
allowed_literal_origins=["http://evil.com"],
177+
allowed_origin_regex=None,
178+
has_configured_allowed_origins=True,
179+
)
180+
assert result, "Explicitly allowed origin should still pass"
181+
182+
# --- Non-loopback server (protection does not apply) ---
183+
184+
def test_non_loopback_server_no_dns_guard(self):
185+
"""Server bound to 0.0.0.0 — DNS guard must not interfere with same-origin check."""
186+
scope = self._make_scope(
187+
server_host="0.0.0.0", host_header="example.com:8000"
188+
)
189+
result = _is_request_origin_allowed(
190+
origin="http://example.com:8000",
191+
scope=scope,
192+
allowed_literal_origins=[],
193+
allowed_origin_regex=None,
194+
has_configured_allowed_origins=False,
195+
)
196+
assert result, "Same-origin on public server should be allowed"

0 commit comments

Comments
 (0)