fix(http): accept IPv6 loopback Host headers

Makefile

@@ -24,7 +24,7 @@ test:
 	@# Note that extensive integration tests are also run in the bundle repo,
 	@# for both aw-server and aw-server-rust, but without code coverage.
 	python -c 'import aw_server'
-	python -m pytest tests/test_server.py tests/test_profile.py tests/test_profile_config.py
+	python -m pytest tests/test_server.py tests/test_profile.py tests/test_profile_config.py tests/test_host_header.py
 
 typecheck:
 	python -m mypy aw_server tests --ignore-missing-imports

README.md

@@ -28,6 +28,10 @@ Run aw-server:
 aw-server
 ```
 
+To bind explicitly to IPv6 loopback, run `aw-server --host ::1` and open
+`http://[::1]:5600`. The default `localhost` binding depends on your system's
+address resolution and may listen only on IPv4.
+
 ## Development
 
 If you want to run aw-server in development, you probably want to run a

aw_server/rest.py

@@ -1,8 +1,10 @@
 import json
+import re
 import traceback
 from functools import wraps
 from threading import Lock
 from typing import Dict
+from urllib.parse import urlsplit
 
 import iso8601
 from aw_core import schema
@@ -40,7 +42,31 @@ def decorator(*args, **kwargs):
         elif req_host is None:
             return {"message": "host header is missing"}, 400
         else:
-            if req_host.split(":")[0] not in ["localhost", "127.0.0.1", server_host]:
+            try:
+                # Host is an authority: IPv6 literals use brackets, optionally
+                # followed by a port. Splitting on ':' truncates them to '['.
+                authority = urlsplit("//" + req_host)
+                # urlsplit accepts empty ports and discards empty query or
+                # fragment delimiters. Require a complete Host authority too.
+                valid = (
+                    re.fullmatch(
+                        r"(?:\[[^\[\]\s/?#@]+\]|[^\s:/?#@\[\]]+)(?::[0-9]+)?",
+                        req_host,
+                    )
+                    is not None
+                    and authority.hostname
+                    in ["localhost", "127.0.0.1", "::1", server_host.lower()]
+                    and authority.username is None
+                    and authority.password is None
+                    and not authority.path
+                    and not authority.query
+                    and not authority.fragment
+                )
+                # Validate the optional port too (including unbracketed IPv6).
+                authority.port
+            except ValueError:
+                valid = False
+            if not valid:
                 return {"message": f"host header is invalid (was {req_host})"}, 400
         return f(*args, **kwargs)
 

tests/test_host_header.py

@@ -0,0 +1,95 @@
+import http.client
+import json
+import socket
+from threading import Thread
+
+import pytest
+from flask import request
+from werkzeug.serving import make_server
+
+from aw_server.server import AWFlask
+
+
+@pytest.mark.parametrize(
+    "host", ["localhost", "localhost:5600", "127.0.0.1:5600", "[::1]", "[::1]:5600"]
+)
+def test_loopback_host_headers(flask_client, host):
+    response = flask_client.get("/api/0/buckets/", headers={"Host": host})
+    assert response.status_code == 200
+
+
+@pytest.mark.parametrize(
+    "host",
+    [
+        "evil.example",
+        "evil.example:5600",
+        "[2001:db8::1]:5600",
+        "::1",
+        "[::1",
+        "[::1]:bad",
+        "[::1]:65536",
+        "evil.example@localhost",
+        "localhost/evil",
+        "localhost?evil",
+        "localhost#evil",
+        "[::1]:",
+        "[::1]?",
+        "[::1]#",
+        "localhost:",
+        "localhost?",
+        "localhost#",
+        "[::1]junk",
+        "[::1]junk:5600",
+        "local\thost",
+    ],
+)
+def test_untrusted_or_malformed_host_headers(app, host):
+    # Inject after context creation: Werkzeug's test client parses Host for
+    # its cookie jar and rejects malformed authorities before dispatch.
+    with app.test_request_context("/api/0/buckets/"):
+        request.environ["HTTP_HOST"] = host
+        response = app.full_dispatch_request()
+        assert response.status_code == 400
+
+
+def test_configured_ipv6_host():
+    app = AWFlask("2001:db8::1", testing=True, cors_origins=[])
+    assert (
+        app.test_client()
+        .get("/api/0/buckets/", headers={"Host": "[2001:db8::1]:5600"})
+        .status_code
+        == 200
+    )
+
+
+def test_missing_host_header(flask_client):
+    assert (
+        flask_client.get(
+            "/api/0/buckets/", environ_overrides={"HTTP_HOST": ""}
+        ).status_code
+        == 400
+    )
+
+
+def test_ipv6_loopback_listener():
+    # Skip only when this machine cannot bind IPv6 loopback at all.
+    try:
+        with socket.socket(socket.AF_INET6, socket.SOCK_STREAM) as probe:
+            probe.bind(("::1", 0))
+    except OSError as error:
+        pytest.skip(f"IPv6 loopback unavailable: {error}")
+    app = AWFlask("::1", testing=True, cors_origins=[])
+    server = make_server("::1", 0, app, threaded=True)
+    thread = Thread(target=server.serve_forever)
+    thread.start()
+    connection = http.client.HTTPConnection("::1", server.server_port, timeout=5)
+    try:
+        connection.request("GET", "/api/0/buckets/")
+        response = connection.getresponse()
+        assert response.status == 200
+        assert json.loads(response.read()) == {}
+    finally:
+        connection.close()
+        server.shutdown()
+        thread.join(timeout=5)
+        server.server_close()