Reject trusted network access from proxies (#52388)

homeassistant/auth/providers/trusted_networks.py

@@ -81,6 +81,17 @@ class TrustedNetworksAuthProvider(AuthProvider):
         """Return trusted users per network."""
         return cast(Dict[IPNetwork, Any], self.config[CONF_TRUSTED_USERS])
 
+    @property
+    def trusted_proxies(self) -> list[IPNetwork]:
+        """Return trusted proxies in the system."""
+        if not self.hass.http:
+            return []
+
+        return [
+            ip_network(trusted_proxy)
+            for trusted_proxy in self.hass.http.trusted_proxies
+        ]
+
     @property
     def support_mfa(self) -> bool:
         """Trusted Networks auth provider does not support MFA."""
@@ -178,6 +189,9 @@ class TrustedNetworksAuthProvider(AuthProvider):
         ):
             raise InvalidAuthError("Not in trusted_networks")
 
+        if any(ip_addr in trusted_proxy for trusted_proxy in self.trusted_proxies):
+            raise InvalidAuthError("Can't allow access from a proxy server")
+
     @callback
     def async_validate_refresh_token(
         self, refresh_token: RefreshToken, remote_ip: str | None = None

homeassistant/components/http/forwarded.py

@@ -129,11 +129,9 @@ def async_setup_forwarded(
             overrides["remote"] = str(forwarded_ip)
             break
         else:
-            _LOGGER.warning(
-                "Request originated directly from a trusted proxy included in X-Forwarded-For: %s, this is likely a miss configuration and will be rejected",
-                forwarded_for_headers,
-            )
-            raise HTTPBadRequest()
+            # If all the IP addresses are from trusted networks, take the left-most.
+            forwarded_for_index = -1
+            overrides["remote"] = str(forwarded_for[-1])
 
         # Handle X-Forwarded-Proto
         forwarded_proto_headers: list[str] = request.headers.getall(

tests/auth/providers/test_trusted_networks.py

@@ -8,7 +8,9 @@ import voluptuous as vol
 from homeassistant import auth
 from homeassistant.auth import auth_store
 from homeassistant.auth.providers import trusted_networks as tn_auth
+from homeassistant.components.http import CONF_TRUSTED_PROXIES, CONF_USE_X_FORWARDED_FOR
 from homeassistant.data_entry_flow import RESULT_TYPE_ABORT, RESULT_TYPE_CREATE_ENTRY
+from homeassistant.setup import async_setup_component
 
 
 @pytest.fixture
@@ -144,6 +146,29 @@ async def test_validate_access(provider):
         provider.async_validate_access(ip_address("2001:db8::ff00:42:8329"))
 
 
+async def test_validate_access_proxy(hass, provider):
+    """Test validate access from trusted networks are blocked from proxy."""
+
+    await async_setup_component(
+        hass,
+        "http",
+        {
+            "http": {
+                CONF_TRUSTED_PROXIES: ["192.168.128.0/31", "fd00::1"],
+                CONF_USE_X_FORWARDED_FOR: True,
+            }
+        },
+    )
+    provider.async_validate_access(ip_address("192.168.128.2"))
+    provider.async_validate_access(ip_address("fd00::2"))
+    with pytest.raises(tn_auth.InvalidAuthError):
+        provider.async_validate_access(ip_address("192.168.128.0"))
+    with pytest.raises(tn_auth.InvalidAuthError):
+        provider.async_validate_access(ip_address("192.168.128.1"))
+    with pytest.raises(tn_auth.InvalidAuthError):
+        provider.async_validate_access(ip_address("fd00::1"))
+
+
 async def test_validate_refresh_token(provider):
     """Verify re-validation of refresh token."""
     with patch.object(provider, "async_validate_access") as mock:

tests/components/http/test_forwarded.py

@@ -43,9 +43,15 @@ async def test_x_forwarded_for_without_trusted_proxy(aiohttp_client, caplog):
 @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", "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.0/24"], "127.0.0.1", "127.0.0.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"),
         (["127.0.0.1"], "255.255.255.255", "255.255.255.255"),
@@ -77,33 +83,6 @@ async def test_x_forwarded_for_with_trusted_proxy(
     assert resp.status == 200
 
 
-@pytest.mark.parametrize(
-    "trusted_proxies,x_forwarded_for",
-    [
-        (
-            ["127.0.0.0/24", "1.1.1.1", "10.10.10.0/24"],
-            "10.10.10.10, 1.1.1.1",
-        ),
-        (["127.0.0.0/24"], "127.0.0.1"),
-    ],
-)
-async def test_x_forwarded_for_from_trusted_proxy_rejected(
-    trusted_proxies, x_forwarded_for, aiohttp_client
-):
-    """Test that we reject forwarded requests from proxy server itself."""
-
-    app = web.Application()
-    app.router.add_get("/", mock_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: x_forwarded_for})
-
-    assert resp.status == 400
-
-
 async def test_x_forwarded_for_disabled_with_proxy(aiohttp_client, caplog):
     """Test that we warn when processing is disabled, but proxy has been detected."""