diff --git a/python/packages/devui/agent_framework_devui/_server.py b/python/packages/devui/agent_framework_devui/_server.py index bc94dee9a21..aa4248967c7 100644 --- a/python/packages/devui/agent_framework_devui/_server.py +++ b/python/packages/devui/agent_framework_devui/_server.py @@ -137,6 +137,24 @@ def _loopback_allowed_hosts(self) -> frozenset[str] | None: return None return self._LOOPBACK_HOSTS + @staticmethod + def _hostname_from_host_header(host_header: str) -> str: + """Return the host name part of a Host header, lowercased and without any port. + + IPv6 literals keep their brackets, so "[::1]:8080" and "[::1]" both yield "[::1]". + Splitting on the first colon instead would turn both into "[", which no allowlist + member matches. + """ + host = host_header.strip().lower() + if host.startswith("["): + # IPv6 literal, optionally followed by ":port". + end = host.find("]") + return host[: end + 1] if end != -1 else host + if host.count(":") == 1: + # "example.com:8080". A bare IPv6 literal holds more than one colon and has no port. + return host.rsplit(":", 1)[0] + return host + @classmethod def _resolve_auth_token(cls, host: str, auth_enabled: bool, auth_token: str | None) -> str | None: """Resolve the active Bearer token. Returns None when auth is disabled.""" @@ -472,7 +490,7 @@ async def auth_middleware(request: Request, call_next: Callable[[Request], Await @app.middleware("http") async def host_header_middleware(request: Request, call_next: Callable[[Request], Awaitable[Any]]) -> Any: host_header = request.headers.get("host", "") - hostname = host_header.split(":", 1)[0].lower() + hostname = self._hostname_from_host_header(host_header) if hostname and hostname not in expected_hosts: return JSONResponse( status_code=400, diff --git a/python/packages/devui/tests/devui/test_server.py b/python/packages/devui/tests/devui/test_server.py index 56055fd137a..c4989491f64 100644 --- a/python/packages/devui/tests/devui/test_server.py +++ b/python/packages/devui/tests/devui/test_server.py @@ -883,6 +883,46 @@ def test_loopback_bind_rejects_non_allowlisted_host_header(monkeypatch): assert ok_localhost.status_code == 200 +@pytest.mark.parametrize( + ("host_header", "expected_hostname"), + [ + pytest.param("127.0.0.1", "127.0.0.1", id="ipv4"), + pytest.param("127.0.0.1:8080", "127.0.0.1", id="ipv4-with-port"), + pytest.param("localhost:8080", "localhost", id="hostname-with-port"), + pytest.param("LOCALHOST:8080", "localhost", id="hostname-is-lowercased"), + pytest.param("[::1]", "[::1]", id="ipv6-literal"), + pytest.param("[::1]:8080", "[::1]", id="ipv6-literal-with-port"), + pytest.param("[::FFFF:127.0.0.1]:8080", "[::ffff:127.0.0.1]", id="ipv6-mapped-ipv4"), + pytest.param("::1", "::1", id="bare-ipv6-has-no-port-to-strip"), + pytest.param("", "", id="empty"), + ], +) +def test_hostname_from_host_header(host_header: str, expected_hostname: str) -> None: + """The Host header parser strips a port only when the host is not an IPv6 literal.""" + assert DevServer._hostname_from_host_header(host_header) == expected_hostname + + +@pytest.mark.parametrize("bind", ["127.0.0.1", "::1", "[::1]"]) +def test_loopback_bind_accepts_ipv6_loopback_host_header(monkeypatch, bind: str) -> None: + """A browser addresses an IPv6 loopback bind as `[::1]` or `[::1]:`. + + Those two forms used to be reduced to `"["` by a split on the first colon, so every + request to an IPv6 loopback bind was answered with `host_not_allowed`. + """ + monkeypatch.delenv("DEVUI_AUTH_TOKEN", raising=False) + + server = DevServer(host=bind, auth_token="s3cret") + app = server.get_app() + + with TestClient(app, base_url="http://127.0.0.1") as client: + for host_header in ("[::1]", "[::1]:8080"): + response = client.get("/health", headers={"Host": host_header}) + assert response.status_code == 200, f"bind {bind} rejected Host {host_header!r}" + + rebound = client.get("/health", headers={"Host": "evil.example:8080"}) + assert rebound.status_code == 400 + + def test_serve_defaults_to_auth_enabled(): """`serve()`'s public signature must default to auth_enabled=True.""" sig = inspect.signature(agent_framework_devui.serve)