diff --git a/Makefile b/Makefile index 2a58b51..3bb1a21 100644 --- a/Makefile +++ b/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 diff --git a/README.md b/README.md index b49630f..c66535c 100644 --- a/README.md +++ b/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 diff --git a/aw_server/rest.py b/aw_server/rest.py index 0b2c320..b2ecc81 100644 --- a/aw_server/rest.py +++ b/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) diff --git a/tests/test_host_header.py b/tests/test_host_header.py new file mode 100644 index 0000000..43b5c77 --- /dev/null +++ b/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()