Fixed #181 -- Improved Host validation for malformed colon values.

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

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,

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

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(