Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 19 additions & 1 deletion python/packages/devui/agent_framework_devui/_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -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."""
Expand Down Expand Up @@ -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,
Expand Down
40 changes: 40 additions & 0 deletions python/packages/devui/tests/devui/test_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -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]:<port>`.

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)
Expand Down
Loading