starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-11-03 20:23:50 -04:00
Code Issues Packages Projects Releases Wiki Activity

Fix incorrect forward header handling (#154793)

Browse Source
This commit is contained in:
Sebastian Faul
2025-10-19 21:26:12 +02:00
committed by GitHub
parent 204ff5d45f
commit 6edafd8965
2 changed files with 174 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/components/http/test_forwarded.py
+143 -47
View File
@@ -90,6 +90,56 @@ async def test_x_forwarded_for_with_trusted_proxy(
assert resp.status == HTTPStatus.OK
@pytest.mark.parametrize(
("trusted_proxies", "x_forwarded_for", "remote"),
[
(
["127.0.0.0/24", "1.1.1.1", "10.10.10.0/24"],
["10.10.10.10", "1.1.1.1"],
"10.10.10.10",
),
(
["127.0.0.0/24", "1.1.1.1"],
["123.123.123.123", "2.2.2.2", "1.1.1.1"],
"2.2.2.2",
),
(["127.0.0.0/24"], ["123.123.123.123", "2.2.2.2", "1.1.1.1"], "1.1.1.1"),
(["127.0.0.1", "1.1.1.1"], ["123.123.123.123", "1.1.1.1"], "123.123.123.123"),
(
["127.0.0.1", "1.1.1.1"],
["123.123.123.123", "2.2.2.2", "1.1.1.1"],
"2.2.2.2",
),
],
)
async def test_x_multiple_forwarded_for_with_trusted_proxy(
trusted_proxies, x_forwarded_for, remote, aiohttp_client: ClientSessionGenerator
) -> None:
"""Test that we get the IP from multiple forwarded for headers."""
async def handler(request):
url = mock_api_client.make_url("/")
assert request.host == f"{url.host}:{url.port}"
assert request.scheme == "http"
assert not request.secure
assert request.remote == remote
return web.Response()
app = web.Application()
app.router.add_get("/", handler)
async_setup_forwarded(
app, True, [ip_network(trusted_proxy) for trusted_proxy in trusted_proxies]
)
mock_api_client = await aiohttp_client(app)
resp = await mock_api_client.get(
"/", headers=[(X_FORWARDED_FOR, addr) for addr in x_forwarded_for]
)
assert resp.status == HTTPStatus.OK
async def test_x_forwarded_for_disabled_with_proxy(
aiohttp_client: ClientSessionGenerator, caplog: pytest.LogCaptureFixture
) -> None:
@@ -176,28 +226,6 @@ async def test_x_forwarded_for_with_malformed_header(
assert "Invalid IP address in X-Forwarded-For" in caplog.text
async def test_x_forwarded_for_with_multiple_headers(
aiohttp_client: ClientSessionGenerator, caplog: pytest.LogCaptureFixture
) -> None:
"""Test that we get a HTTP 400 bad request with multiple headers."""
app = web.Application()
app.router.add_get("/", mock_handler)
async_setup_forwarded(app, True, [ip_network("127.0.0.1")])
mock_api_client = await aiohttp_client(app)
resp = await mock_api_client.get(
"/",
headers=[
(X_FORWARDED_FOR, "222.222.222.222"),
(X_FORWARDED_FOR, "123.123.123.123"),
],
)
assert resp.status == HTTPStatus.BAD_REQUEST
assert "Too many headers for X-Forwarded-For" in caplog.text
@pytest.mark.parametrize(
("x_forwarded_for", "remote", "x_forwarded_proto", "secure"),
[
@@ -258,6 +286,65 @@ async def test_x_forwarded_proto_with_trusted_proxy(
assert resp.status == HTTPStatus.OK
@pytest.mark.parametrize(
("x_forwarded_for", "remote", "x_forwarded_proto", "secure"),
[
(
"10.10.10.10, 127.0.0.1, 127.0.0.2",
"10.10.10.10",
["https", "http", "http"],
True,
),
(
"10.10.10.10, 127.0.0.1, 127.0.0.2",
"10.10.10.10",
["http", "https", "https"],
False,
),
(
"255.255.255.255, 10.10.10.10, 127.0.0.1",
"10.10.10.10",
["http", "https", "http"],
True,
),
(
"255.255.255.255, 10.10.10.10, 127.0.0.1",
"10.10.10.10",
["https", "http", "https"],
False,
),
],
)
async def test_x_multiple_forwarded_proto_with_trusted_proxy(
x_forwarded_for,
remote,
x_forwarded_proto,
secure,
aiohttp_client: ClientSessionGenerator,
) -> None:
"""Test that we get the proto header if proxy is trusted."""
async def handler(request):
assert request.remote == remote
assert request.scheme == ("https" if secure else "http")
assert request.secure == secure
return web.Response()
app = web.Application()
app.router.add_get("/", handler)
async_setup_forwarded(app, True, [ip_network("127.0.0.0/24")])
mock_api_client = await aiohttp_client(app)
resp = await mock_api_client.get(
"/",
headers=[(X_FORWARDED_FOR, x_forwarded_for)]
+ [(X_FORWARDED_PROTO, proto) for proto in x_forwarded_proto],
)
assert resp.status == HTTPStatus.OK
async def test_x_forwarded_proto_with_trusted_proxy_multiple_for(
aiohttp_client: ClientSessionGenerator,
) -> None:
@@ -288,6 +375,38 @@ async def test_x_forwarded_proto_with_trusted_proxy_multiple_for(
assert resp.status == HTTPStatus.OK
async def test_x_forwarded_proto_with_trusted_proxy_multiple_for_2(
aiohttp_client: ClientSessionGenerator,
) -> None:
"""Test that we get the proto with 1 element in the proto, multiple in the for."""
async def handler(request):
url = mock_api_client.make_url("/")
assert request.host == f"{url.host}:{url.port}"
assert request.scheme == "https"
assert request.secure
assert request.remote == "255.255.255.255"
return web.Response()
app = web.Application()
app.router.add_get("/", handler)
async_setup_forwarded(app, True, [ip_network("127.0.0.0/24")])
mock_api_client = await aiohttp_client(app)
resp = await mock_api_client.get(
"/",
headers=[
(X_FORWARDED_FOR, "255.255.255.255"),
(X_FORWARDED_FOR, "127.0.0.1"),
(X_FORWARDED_FOR, "127.0.0.2"),
(X_FORWARDED_PROTO, "https"),
],
)
assert resp.status == HTTPStatus.OK
async def test_x_forwarded_proto_not_processed_without_for(
aiohttp_client: ClientSessionGenerator,
) -> None:
@@ -312,28 +431,6 @@ async def test_x_forwarded_proto_not_processed_without_for(
assert resp.status == HTTPStatus.OK
async def test_x_forwarded_proto_with_multiple_headers(
aiohttp_client: ClientSessionGenerator, caplog: pytest.LogCaptureFixture
) -> None:
"""Test that we get a HTTP 400 bad request with multiple headers."""
app = web.Application()
app.router.add_get("/", mock_handler)
async_setup_forwarded(app, True, [ip_network("127.0.0.1")])
mock_api_client = await aiohttp_client(app)
resp = await mock_api_client.get(
"/",
headers=[
(X_FORWARDED_FOR, "222.222.222.222"),
(X_FORWARDED_PROTO, "https"),
(X_FORWARDED_PROTO, "http"),
],
)
assert resp.status == HTTPStatus.BAD_REQUEST
assert "Too many headers for X-Forward-Proto" in caplog.text
@pytest.mark.parametrize(
"x_forwarded_proto",
["", ",", "https, , https", "https, https, "],
@@ -447,7 +544,7 @@ async def test_x_forwarded_host_not_processed_without_for(
async def test_x_forwarded_host_with_multiple_headers(
aiohttp_client: ClientSessionGenerator, caplog: pytest.LogCaptureFixture
) -> None:
"""Test that we get a HTTP 400 bad request with multiple headers."""
"""Test that we get a HTTP 200 OK with multiple headers."""
app = web.Application()
app.router.add_get("/", mock_handler)
async_setup_forwarded(app, True, [ip_network("127.0.0.1")])
@@ -462,8 +559,7 @@ async def test_x_forwarded_host_with_multiple_headers(
],
)
assert resp.status == HTTPStatus.BAD_REQUEST
assert "Too many headers for X-Forwarded-Host" in caplog.text
assert resp.status == HTTPStatus.OK
async def test_x_forwarded_host_with_empty_header(