evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Trusted networks auth provider warns if detects a requests with x-forwarded-for header while the http integration is not configured for reverse proxies (#51319)

Browse source 
* Trusted networks auth provider to require http integration configured for proxies to allow logging in with requests with x-forwarded-for header

* Make it a warning
This commit is contained in:
Paulus Schoutsen 2021-06-01 03:51:44 -07:00 • committed by GitHub
parent 94ae8396dd
commit bd0373388d
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 146 additions and 40 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

111
homeassistant/auth/providers/trusted_networks.py
View file

@ -14,10 +14,13 @@ from ipaddress import (
ip_address,
ip_network,
)
import logging
from typing import Any, Dict, List, Union, cast
from aiohttp import hdrs
import voluptuous as vol
from homeassistant.components import http
from homeassistant.core import callback
from homeassistant.data_entry_flow import FlowResult
from homeassistant.exceptions import HomeAssistantError
@ -86,40 +89,60 @@ class TrustedNetworksAuthProvider(AuthProvider):
"""Trusted Networks auth provider does not support MFA."""
return False
@callback
def is_allowed_request(self) -> bool:
"""Return if it is an allowed request."""
request = http.current_request.get()
if request is not None and (
self.hass.http.use_x_forwarded_for
or hdrs.X_FORWARDED_FOR not in request.headers
):
return True
logging.getLogger(__name__).warning(
"A request contained an x-forwarded-for header but your HTTP integration is not set-up "
"for reverse proxies. This usually means that you have not configured your reverse proxy "
"correctly. This request will be blocked in Home Assistant 2021.7 unless you configure "
"your HTTP integration to allow this header."
)
return True
async def async_login_flow(self, context: dict | None) -> LoginFlow:
"""Return a flow to login."""
assert context is not None
if not self.is_allowed_request():
return MisconfiguredTrustedNetworksLoginFlow(self)
ip_addr = cast(IPAddress, context.get("ip_address"))
users = await self.store.async_get_users()
available_users = [
user for user in users if not user.system_generated and user.is_active
]
for ip_net, user_or_group_list in self.trusted_users.items():
if ip_addr in ip_net:
user_list = [
user_id
for user_id in user_or_group_list
if isinstance(user_id, str)
]
group_list = [
group[CONF_GROUP]
for group in user_or_group_list
if isinstance(group, dict)
]
flattened_group_list = [
group for sublist in group_list for group in sublist
]
available_users = [
user
for user in available_users
if (
user.id in user_list
or any(
group.id in flattened_group_list for group in user.groups
)
)
]
break
if ip_addr not in ip_net:
continue
user_list = [
user_id for user_id in user_or_group_list if isinstance(user_id, str)
]
group_list = [
group[CONF_GROUP]
for group in user_or_group_list
if isinstance(group, dict)
]
flattened_group_list = [
group for sublist in group_list for group in sublist
]
available_users = [
user
for user in available_users
if (
user.id in user_list
or any(group.id in flattened_group_list for group in user.groups)
)
]
break
return TrustedNetworksLoginFlow(
self,
@ -136,13 +159,22 @@ class TrustedNetworksAuthProvider(AuthProvider):
users = await self.store.async_get_users()
for user in users:
if not user.system_generated and user.is_active and user.id == user_id:
for credential in await self.async_credentials():
if credential.data["user_id"] == user_id:
return credential
cred = self.async_create_credentials({"user_id": user_id})
await self.store.async_link_user(user, cred)
return cred
if user.id != user_id:
continue
if user.system_generated:
continue
if not user.is_active:
continue
for credential in await self.async_credentials():
if credential.data["user_id"] == user_id:
return credential
cred = self.async_create_credentials({"user_id": user_id})
await self.store.async_link_user(user, cred)
return cred
# We only allow login as exist user
raise InvalidUserError
@ -163,6 +195,11 @@ class TrustedNetworksAuthProvider(AuthProvider):
Raise InvalidAuthError if not.
Raise InvalidAuthError if trusted_networks is not configured.
"""
if not self.is_allowed_request():
raise InvalidAuthError(
"No request or it contains x-forwarded-for header and that's not allowed by configuration"
)
if not self.trusted_networks:
raise InvalidAuthError("trusted_networks is not configured")
@ -183,6 +220,16 @@ class TrustedNetworksAuthProvider(AuthProvider):
self.async_validate_access(ip_address(remote_ip))
class MisconfiguredTrustedNetworksLoginFlow(LoginFlow):
"""Login handler for misconfigured trusted networks."""
async def async_step_init(
self, user_input: dict[str, str] | None = None
) -> FlowResult:
"""Handle the step of the form."""
return self.async_abort(reason="forwared_for_header_not_allowed")
class TrustedNetworksLoginFlow(LoginFlow):
"""Handler for the login flow."""

1
homeassistant/components/http/__init__.py
View file

@ -252,6 +252,7 @@ class HomeAssistantHTTP:
self.ssl_key = ssl_key
self.server_host = server_host
self.server_port = server_port
self.use_x_forwarded_for = use_x_forwarded_for
self.trusted_proxies = trusted_proxies
self.is_ban_enabled = is_ban_enabled
self.ssl_profile = ssl_profile

74
tests/auth/providers/test_trusted_networks.py
View file

@ -5,10 +5,13 @@ from unittest.mock import Mock, patch
import pytest
import voluptuous as vol
from homeassistant import auth
from homeassistant import auth, const
from homeassistant.auth import auth_store
from homeassistant.auth.providers import trusted_networks as tn_auth
from homeassistant.data_entry_flow import RESULT_TYPE_ABORT, RESULT_TYPE_CREATE_ENTRY
from homeassistant.setup import async_setup_component
FORWARD_FOR_IS_WARNING = (const.MAJOR_VERSION, const.MINOR_VERSION) < (2021, 8)
@pytest.fixture
@ -112,7 +115,17 @@ def manager_bypass_login(hass, store, provider_bypass_login):
)
async def test_trusted_networks_credentials(manager, provider):
@pytest.fixture
def mock_allowed_request():
"""Mock that the request is allowed."""
with patch(
"homeassistant.auth.providers.trusted_networks.TrustedNetworksAuthProvider.is_allowed_request",
return_value=True,
):
yield
async def test_trusted_networks_credentials(manager, provider, mock_allowed_request):
"""Test trusted_networks credentials related functions."""
owner = await manager.async_create_user("test-owner")
tn_owner_cred = await provider.async_get_or_create_credentials({"user": owner.id})
@ -129,7 +142,7 @@ async def test_trusted_networks_credentials(manager, provider):
await provider.async_get_or_create_credentials({"user": "invalid-user"})
async def test_validate_access(provider):
async def test_validate_access(provider, mock_allowed_request):
"""Test validate access from trusted networks."""
provider.async_validate_access(ip_address("192.168.0.1"))
provider.async_validate_access(ip_address("192.168.128.10"))
@ -144,7 +157,7 @@ async def test_validate_access(provider):
provider.async_validate_access(ip_address("2001:db8::ff00:42:8329"))
async def test_validate_refresh_token(provider):
async def test_validate_refresh_token(provider, mock_allowed_request):
"""Verify re-validation of refresh token."""
with patch.object(provider, "async_validate_access") as mock:
with pytest.raises(tn_auth.InvalidAuthError):
@ -154,7 +167,7 @@ async def test_validate_refresh_token(provider):
mock.assert_called_once_with(ip_address("127.0.0.1"))
async def test_login_flow(manager, provider):
async def test_login_flow(manager, provider, mock_allowed_request):
"""Test login flow."""
owner = await manager.async_create_user("test-owner")
user = await manager.async_create_user("test-user")
@ -181,7 +194,9 @@ async def test_login_flow(manager, provider):
assert step["data"]["user"] == user.id
async def test_trusted_users_login(manager_with_user, provider_with_user):
async def test_trusted_users_login(
manager_with_user, provider_with_user, mock_allowed_request
):
"""Test available user list changed per different IP."""
owner = await manager_with_user.async_create_user("test-owner")
sys_user = await manager_with_user.async_create_system_user(
@ -261,7 +276,9 @@ async def test_trusted_users_login(manager_with_user, provider_with_user):
assert schema({"user": sys_user.id})
async def test_trusted_group_login(manager_with_user, provider_with_user):
async def test_trusted_group_login(
manager_with_user, provider_with_user, mock_allowed_request
):
"""Test config trusted_user with group_id."""
owner = await manager_with_user.async_create_user("test-owner")
# create a user in user group
@ -314,7 +331,9 @@ async def test_trusted_group_login(manager_with_user, provider_with_user):
assert schema({"user": user.id})
async def test_bypass_login_flow(manager_bypass_login, provider_bypass_login):
async def test_bypass_login_flow(
manager_bypass_login, provider_bypass_login, mock_allowed_request
):
"""Test login flow can be bypass if only one user available."""
owner = await manager_bypass_login.async_create_user("test-owner")
@ -345,3 +364,42 @@ async def test_bypass_login_flow(manager_bypass_login, provider_bypass_login):
# both owner and user listed
assert schema({"user": owner.id})
assert schema({"user": user.id})
async def test_allowed_request(hass, provider, current_request, caplog):
"""Test allowing requests."""
assert await async_setup_component(hass, "http", {})
provider.async_validate_access(ip_address("192.168.0.1"))
current_request.get.return_value = current_request.get.return_value.clone(
headers={
**current_request.get.return_value.headers,
"x-forwarded-for": "1.2.3.4",
}
)
if FORWARD_FOR_IS_WARNING:
caplog.clear()
provider.async_validate_access(ip_address("192.168.0.1"))
assert "This request will be blocked" in caplog.text
else:
with pytest.raises(tn_auth.InvalidAuthError):
provider.async_validate_access(ip_address("192.168.0.1"))
hass.http.use_x_forwarded_for = True
provider.async_validate_access(ip_address("192.168.0.1"))
@pytest.mark.skipif(FORWARD_FOR_IS_WARNING, reason="Currently a warning")
async def test_login_flow_no_request(provider):
"""Test getting a login flow."""
login_flow = await provider.async_login_flow({"ip_address": ip_address("1.1.1.1")})
assert await login_flow.async_step_init() == {
"description_placeholders": None,
"flow_id": None,
"handler": None,
"reason": "forwared_for_header_not_allowed",
"type": "abort",
}