diff --git a/docs/en/dev/changelog.md b/docs/en/dev/changelog.md index 539eaba..c36f1ce 100644 --- a/docs/en/dev/changelog.md +++ b/docs/en/dev/changelog.md @@ -37,6 +37,12 @@ cut. ### Fixed +- Host header validation now rejects malformed authority values before auth or + handlers run. Bad ports such as `localhost:bad` and bare IPv6 values such as + `::1` now return `400 Invalid Host header`, even when `allowed_hosts=["*"]`; + valid bracketed IPv6 hosts such as `[::1]` and `[::1]:8000` still work. + ([#181](https://github.com/DevilsAutumn/quater/issues/181)) + - Fixed resource provider planning so valid provider parameters still resolve when the same provider has an unrelated broken return annotation. This covers `Annotated[..., Resource]` dependencies and `Request` parameters that are not diff --git a/src/quater/security.py b/src/quater/security.py index 0df37d6..13838a6 100644 --- a/src/quater/security.py +++ b/src/quater/security.py @@ -198,7 +198,7 @@ def _validate_allowed_host( def _host_matches(host: str, allowed: str) -> bool: - normalized_allowed = _normalize_host(allowed) + normalized_allowed = _normalize_allowed_host(allowed) if normalized_allowed == "*": return True if normalized_allowed is None: @@ -241,30 +241,59 @@ def _normalize_host(value: str | None) -> str | None: if not host: return None if host.startswith("["): - end = host.find("]") - if end == -1: - return None - remainder = host[end + 1 :] - if remainder and (not remainder.startswith(":") or not remainder[1:].isdigit()): - return None - literal = host[1:end] - try: - parsed = ip_address(literal) - except ValueError: + return _normalize_bracketed_ipv6_host(host) + + if ":" in host: + if host.count(":") != 1: return None - if parsed.version != 6: + name, port = host.rsplit(":", 1) + if not name or not _is_ascii_digits(port): return None - return literal + host = name + if host.endswith("."): host = host[:-1] - - if host.count(":") == 1: - name, port = host.rsplit(":", 1) - if port.isdigit(): - return name + if not host: + return None return host +def _normalize_allowed_host(value: str) -> str | None: + normalized = _normalize_host(value) + if normalized is not None: + return normalized + + host = value.strip().lower() + try: + parsed = ip_address(host) + except ValueError: + return None + return str(parsed) + + +def _normalize_bracketed_ipv6_host(host: str) -> str | None: + end = host.find("]") + if end == -1: + return None + remainder = host[end + 1 :] + if remainder and ( + not remainder.startswith(":") or not _is_ascii_digits(remainder[1:]) + ): + return None + literal = host[1:end] + try: + parsed = ip_address(literal) + except ValueError: + return None + if parsed.version != 6: + return None + return str(parsed) + + +def _is_ascii_digits(value: str) -> bool: + return bool(value) and value.isascii() and value.isdigit() + + def _set_default_header( headers: tuple[tuple[str, str], ...], name: str, diff --git a/tests/integration/test_cross_transport_behavior.py b/tests/integration/test_cross_transport_behavior.py index 140227e..a678b8f 100644 --- a/tests/integration/test_cross_transport_behavior.py +++ b/tests/integration/test_cross_transport_behavior.py @@ -543,3 +543,39 @@ async def test_allowed_host_rejection_is_transport_independent() -> None: assert headers["x-content-type-options"] == "nosniff" assert auth_subjects == [] assert handler_calls == [] + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("host", "allowed_hosts"), + ( + ("localhost:bad", ("*",)), + ("::1", None), + ), + ids=("bad-port-with-wildcard", "bare-ipv6-default-local"), +) +async def test_malformed_host_syntax_rejection_is_transport_independent( + host: str, + allowed_hosts: tuple[str, ...] | None, +) -> None: + handler_calls = 0 + + app = Quater() if allowed_hosts is None else Quater(allowed_hosts=allowed_hosts) + + @app.get("/me") + async def me() -> dict[str, bool]: + nonlocal handler_calls + handler_calls += 1 + return {"ok": True} + + responses = [ + await asgi_response(app, host=host, authorization="user_1"), + wsgi_response(app, host=host, authorization="user_1"), + await rsgi_response(app, host=host, authorization="user_1"), + ] + + for status, headers, body in responses: + assert status == 400 + assert body == b"Invalid Host header" + assert headers["x-content-type-options"] == "nosniff" + assert handler_calls == 0 diff --git a/tests/unit/test_allowed_hosts.py b/tests/unit/test_allowed_hosts.py index 5667803..e88a002 100644 --- a/tests/unit/test_allowed_hosts.py +++ b/tests/unit/test_allowed_hosts.py @@ -1,5 +1,7 @@ from __future__ import annotations +from typing import Literal + import pytest from quater import AuthConfig, Quater, Request @@ -35,6 +37,8 @@ async def health() -> dict[str, bool]: "[api.example.com]evil.test", "[api.example.com]:bad", "[api.example.com]:", + "[::1", + "[127.0.0.1]", ), ) async def test_allowed_host_rejects_malformed_bracketed_hosts(host: str) -> None: @@ -67,6 +71,137 @@ async def private() -> dict[str, bool]: assert handler_calls == 0 +@pytest.mark.asyncio +@pytest.mark.parametrize( + "host", + ( + "localhost:bad", + "localhost:", + "api.example.com:bad", + "127.0.0.1:bad", + "example.com:443abc", + "localhost:\xb2", + ), +) +@pytest.mark.parametrize("mode", ("wildcard", "relaxed", "off")) +async def test_malformed_unbracketed_host_is_rejected_before_auth_or_handler( + host: str, + mode: Literal["wildcard", "relaxed", "off"], +) -> None: + auth_calls = 0 + handler_calls = 0 + + async def authenticate(ctx: Request) -> AuthContext | None: + nonlocal auth_calls + auth_calls += 1 + return AuthContext(subject="user_1") + + if mode == "wildcard": + app = Quater( + allowed_hosts=["*"], + auth=[AuthConfig(authenticate, surfaces=["api"])], + ) + else: + app = Quater( + security=mode, + auth=[AuthConfig(authenticate, surfaces=["api"])], + ) + + @app.get("/private") + async def private() -> dict[str, bool]: + nonlocal handler_calls + handler_calls += 1 + return {"ok": True} + + response = await app.handle( + Request(method="GET", path="/private", headers={"host": host}) + ) + + assert response.status_code == 400 + assert response.body == b"Invalid Host header" + assert auth_calls == 0 + assert handler_calls == 0 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("host", ("[::1]:bad", "[::1]:", "[::1]:\xb2")) +async def test_bracketed_ipv6_rejects_malformed_port_with_wildcard( + host: str, +) -> None: + auth_calls = 0 + handler_calls = 0 + + async def authenticate(ctx: Request) -> AuthContext | None: + nonlocal auth_calls + auth_calls += 1 + return AuthContext(subject="user_1") + + app = Quater( + allowed_hosts=["*"], + auth=[AuthConfig(authenticate, surfaces=["api"])], + ) + + @app.get("/private") + async def private() -> dict[str, bool]: + nonlocal handler_calls + handler_calls += 1 + return {"ok": True} + + response = await app.handle( + Request(method="GET", path="/private", headers={"host": host}) + ) + + assert response.status_code == 400 + assert response.body == b"Invalid Host header" + assert auth_calls == 0 + assert handler_calls == 0 + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("host", "allowed_hosts"), + ( + ("::1", None), + ("::1", ("::1",)), + ("2001:db8::1", ("*",)), + ), +) +async def test_bare_ipv6_host_is_rejected_before_auth_or_handler( + host: str, + allowed_hosts: tuple[str, ...] | None, +) -> None: + auth_calls = 0 + handler_calls = 0 + + async def authenticate(ctx: Request) -> AuthContext | None: + nonlocal auth_calls + auth_calls += 1 + return AuthContext(subject="user_1") + + if allowed_hosts is None: + app = Quater(auth=[AuthConfig(authenticate, surfaces=["api"])]) + else: + app = Quater( + allowed_hosts=allowed_hosts, + auth=[AuthConfig(authenticate, surfaces=["api"])], + ) + + @app.get("/private") + async def private() -> dict[str, bool]: + nonlocal handler_calls + handler_calls += 1 + return {"ok": True} + + response = await app.handle( + Request(method="GET", path="/private", headers={"host": host}) + ) + + assert response.status_code == 400 + assert response.body == b"Invalid Host header" + assert auth_calls == 0 + assert handler_calls == 0 + + @pytest.mark.asyncio @pytest.mark.parametrize("host", ("[::1]", "[::1]:8443")) async def test_allowed_host_accepts_bracketed_ipv6_hosts(host: str) -> None: @@ -85,7 +220,7 @@ async def health() -> dict[str, bool]: @pytest.mark.asyncio -@pytest.mark.parametrize("host", ("[localhost]", "[::1].")) +@pytest.mark.parametrize("host", ("[localhost]", "[::1].", ".")) async def test_present_malformed_host_is_rejected_in_strict_default_mode( host: str, ) -> None: @@ -111,6 +246,7 @@ async def health() -> dict[str, bool]: "localhost:8000", "127.0.0.1", "127.0.0.1:8000", + "[::1]", "[::1]:8000", "testserver", ), @@ -308,6 +444,72 @@ async def health() -> dict[str, bool]: assert response.body == b"Invalid Host header" +@pytest.mark.asyncio +async def test_bare_ipv6_authority_is_rejected_even_when_host_is_bracketed() -> None: + auth_calls = 0 + handler_calls = 0 + + async def authenticate(ctx: Request) -> AuthContext | None: + nonlocal auth_calls + auth_calls += 1 + return AuthContext(subject="user_1") + + app = Quater( + allowed_hosts=["::1"], + auth=[AuthConfig(authenticate, surfaces=["api"])], + ) + + @app.get("/private") + async def private() -> dict[str, bool]: + nonlocal handler_calls + handler_calls += 1 + return {"ok": True} + + response = await app.handle( + Request( + method="GET", + path="/private", + headers=(("host", "[::1]"), (":authority", "::1")), + ) + ) + + assert response.status_code == 400 + assert response.body == b"Invalid Host header" + assert auth_calls == 0 + assert handler_calls == 0 + + +@pytest.mark.asyncio +async def test_malformed_allowed_host_pattern_does_not_match_valid_host() -> None: + auth_calls = 0 + handler_calls = 0 + + async def authenticate(ctx: Request) -> AuthContext | None: + nonlocal auth_calls + auth_calls += 1 + return AuthContext(subject="user_1") + + app = Quater( + allowed_hosts=["localhost:bad"], + auth=[AuthConfig(authenticate, surfaces=["api"])], + ) + + @app.get("/private") + async def private() -> dict[str, bool]: + nonlocal handler_calls + handler_calls += 1 + return {"ok": True} + + response = await app.handle( + Request(method="GET", path="/private", headers={"host": "localhost"}) + ) + + assert response.status_code == 400 + assert response.body == b"Invalid Host header" + assert auth_calls == 0 + assert handler_calls == 0 + + @pytest.mark.asyncio async def test_forwarded_host_is_honored_only_from_trusted_proxy() -> None: app = Quater(