mirror of
https://github.com/home-assistant/core.git
synced 2025-04-23 16:57:53 +00:00
Strip login username in backend (#20150)
* Add modern mode to HA auth provider that strips usernames * Add tests for async_get_or_create_credentials * Fix test
This commit is contained in:
Paulus Schoutsen
Jason Hu
parent
9bb7e40ee3
commit
84a2e5d8fb
@ -1,6 +1,7 @@
|
||||
"""Home Assistant auth provider."""
|
||||
import base64
|
||||
from collections import OrderedDict
|
||||
import logging
|
||||
from typing import Any, Dict, List, Optional, cast
|
||||
|
||||
import bcrypt
|
||||
@ -51,6 +52,15 @@ class Data:
|
||||
self._store = hass.helpers.storage.Store(STORAGE_VERSION, STORAGE_KEY,
|
||||
private=True)
|
||||
self._data = None # type: Optional[Dict[str, Any]]
|
||||
self.is_legacy = False
|
||||
|
||||
@callback
|
||||
def normalize_username(self, username: str) -> str:
|
||||
"""Normalize a username based on the mode."""
|
||||
if self.is_legacy:
|
||||
return username
|
||||
|
||||
return username.strip()
|
||||
|
||||
async def async_load(self) -> None:
|
||||
"""Load stored data."""
|
||||
@ -61,6 +71,20 @@ class Data:
|
||||
'users': []
|
||||
}
|
||||
|
||||
for user in data['users']:
|
||||
username = user['username']
|
||||
|
||||
# check if we have unstripped usernames
|
||||
if username != username.strip():
|
||||
self.is_legacy = True
|
||||
|
||||
logging.getLogger(__name__).warning(
|
||||
"Home Assistant auth provider is running in legacy mode "
|
||||
"because we detected usernames that start or end in a "
|
||||
"space. Please change the username.")
|
||||
|
||||
break
|
||||
|
||||
self._data = data
|
||||
|
||||
@property
|
||||
@ -73,6 +97,7 @@ class Data:
|
||||
|
||||
Raises InvalidAuth if auth invalid.
|
||||
"""
|
||||
username = self.normalize_username(username)
|
||||
dummy = b'$2b$12$CiuFGszHx9eNHxPuQcwBWez4CwDTOcLTX5CbOpV6gef2nYuXkY7BO'
|
||||
found = None
|
||||
|
||||
@ -105,7 +130,10 @@ class Data:
|
||||
|
||||
def add_auth(self, username: str, password: str) -> None:
|
||||
"""Add a new authenticated user/pass."""
|
||||
if any(user['username'] == username for user in self.users):
|
||||
username = self.normalize_username(username)
|
||||
|
||||
if any(self.normalize_username(user['username']) == username
|
||||
for user in self.users):
|
||||
raise InvalidUser
|
||||
|
||||
self.users.append({
|
||||
@ -116,9 +144,11 @@ class Data:
|
||||
@callback
|
||||
def async_remove_auth(self, username: str) -> None:
|
||||
"""Remove authentication."""
|
||||
username = self.normalize_username(username)
|
||||
|
||||
index = None
|
||||
for i, user in enumerate(self.users):
|
||||
if user['username'] == username:
|
||||
if self.normalize_username(user['username']) == username:
|
||||
index = i
|
||||
break
|
||||
|
||||
@ -132,8 +162,10 @@ class Data:
|
||||
|
||||
Raises InvalidUser if user cannot be found.
|
||||
"""
|
||||
username = self.normalize_username(username)
|
||||
|
||||
for user in self.users:
|
||||
if user['username'] == username:
|
||||
if self.normalize_username(user['username']) == username:
|
||||
user['password'] = self.hash_password(
|
||||
new_password, True).decode()
|
||||
break
|
||||
@ -178,10 +210,15 @@ class HassAuthProvider(AuthProvider):
|
||||
async def async_get_or_create_credentials(
|
||||
self, flow_result: Dict[str, str]) -> Credentials:
|
||||
"""Get credentials based on the flow result."""
|
||||
username = flow_result['username']
|
||||
if self.data is None:
|
||||
await self.async_initialize()
|
||||
assert self.data is not None
|
||||
|
||||
norm_username = self.data.normalize_username
|
||||
username = norm_username(flow_result['username'])
|
||||
|
||||
for credential in await self.async_credentials():
|
||||
if credential.data['username'] == username:
|
||||
if norm_username(credential.data['username']) == username:
|
||||
return credential
|
||||
|
||||
# Create new credentials.
|
||||
|
||||
@ -1,5 +1,5 @@
|
||||
"""Test the Home Assistant local auth provider."""
|
||||
from unittest.mock import Mock
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
import pytest
|
||||
import voluptuous as vol
|
||||
@ -9,6 +9,8 @@ from homeassistant.auth import auth_manager_from_config, auth_store
|
||||
from homeassistant.auth.providers import (
|
||||
auth_provider_from_config, homeassistant as hass_auth)
|
||||
|
||||
from tests.common import mock_coro
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def data(hass):
|
||||
@ -18,17 +20,13 @@ def data(hass):
|
||||
return data
|
||||
|
||||
|
||||
async def test_adding_user(data, hass):
|
||||
"""Test adding a user."""
|
||||
data.add_auth('test-user', 'test-pass')
|
||||
data.validate_login('test-user', 'test-pass')
|
||||
|
||||
|
||||
async def test_adding_user_duplicate_username(data, hass):
|
||||
"""Test adding a user with duplicate username."""
|
||||
data.add_auth('test-user', 'test-pass')
|
||||
with pytest.raises(hass_auth.InvalidUser):
|
||||
data.add_auth('test-user', 'other-pass')
|
||||
@pytest.fixture
|
||||
def legacy_data(hass):
|
||||
"""Create a loaded legacy data class."""
|
||||
data = hass_auth.Data(hass)
|
||||
hass.loop.run_until_complete(data.async_load())
|
||||
data.is_legacy = True
|
||||
return data
|
||||
|
||||
|
||||
async def test_validating_password_invalid_user(data, hass):
|
||||
@ -37,30 +35,72 @@ async def test_validating_password_invalid_user(data, hass):
|
||||
data.validate_login('non-existing', 'pw')
|
||||
|
||||
|
||||
async def test_not_allow_set_id():
|
||||
"""Test we are not allowed to set an ID in config."""
|
||||
hass = Mock()
|
||||
with pytest.raises(vol.Invalid):
|
||||
await auth_provider_from_config(hass, None, {
|
||||
'type': 'homeassistant',
|
||||
'id': 'invalid',
|
||||
})
|
||||
|
||||
|
||||
async def test_new_users_populate_values(hass, data):
|
||||
"""Test that we populate data for new users."""
|
||||
data.add_auth('hello', 'test-pass')
|
||||
await data.async_save()
|
||||
|
||||
manager = await auth_manager_from_config(hass, [{
|
||||
'type': 'homeassistant'
|
||||
}], [])
|
||||
provider = manager.auth_providers[0]
|
||||
credentials = await provider.async_get_or_create_credentials({
|
||||
'username': 'hello'
|
||||
})
|
||||
user = await manager.async_get_or_create_user(credentials)
|
||||
assert user.name == 'hello'
|
||||
assert user.is_active
|
||||
|
||||
|
||||
async def test_changing_password_raises_invalid_user(data, hass):
|
||||
"""Test that changing password raises invalid user."""
|
||||
with pytest.raises(hass_auth.InvalidUser):
|
||||
data.change_password('non-existing', 'pw')
|
||||
|
||||
|
||||
# Modern mode
|
||||
|
||||
async def test_adding_user(data, hass):
|
||||
"""Test adding a user."""
|
||||
data.add_auth('test-user', 'test-pass')
|
||||
data.validate_login('test-user', 'test-pass')
|
||||
data.validate_login(' test-user ', 'test-pass')
|
||||
|
||||
|
||||
async def test_adding_user_duplicate_username(data, hass):
|
||||
"""Test adding a user with duplicate username."""
|
||||
data.add_auth('test-user', 'test-pass')
|
||||
with pytest.raises(hass_auth.InvalidUser):
|
||||
data.add_auth('test-user ', 'other-pass')
|
||||
|
||||
|
||||
async def test_validating_password_invalid_password(data, hass):
|
||||
"""Test validating an invalid password."""
|
||||
data.add_auth('test-user', 'test-pass')
|
||||
|
||||
with pytest.raises(hass_auth.InvalidAuth):
|
||||
data.validate_login('test-user', 'invalid-pass')
|
||||
data.validate_login(' test-user ', 'invalid-pass')
|
||||
|
||||
|
||||
async def test_changing_password(data, hass):
|
||||
"""Test adding a user."""
|
||||
user = 'test-user'
|
||||
data.add_auth(user, 'test-pass')
|
||||
data.change_password(user, 'new-pass')
|
||||
data.add_auth('test-user', 'test-pass')
|
||||
data.change_password('test-user ', 'new-pass')
|
||||
|
||||
with pytest.raises(hass_auth.InvalidAuth):
|
||||
data.validate_login(user, 'test-pass')
|
||||
data.validate_login('test-user', 'test-pass')
|
||||
|
||||
data.validate_login(user, 'new-pass')
|
||||
|
||||
|
||||
async def test_changing_password_raises_invalid_user(data, hass):
|
||||
"""Test that we initialize an empty config."""
|
||||
with pytest.raises(hass_auth.InvalidUser):
|
||||
data.change_password('non-existing', 'pw')
|
||||
data.validate_login('test-user', 'new-pass')
|
||||
|
||||
|
||||
async def test_login_flow_validates(data, hass):
|
||||
@ -81,6 +121,110 @@ async def test_login_flow_validates(data, hass):
|
||||
assert result['type'] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result['errors']['base'] == 'invalid_auth'
|
||||
|
||||
result = await flow.async_step_init({
|
||||
'username': 'test-user ',
|
||||
'password': 'incorrect-pass',
|
||||
})
|
||||
assert result['type'] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result['errors']['base'] == 'invalid_auth'
|
||||
|
||||
result = await flow.async_step_init({
|
||||
'username': 'test-user',
|
||||
'password': 'test-pass',
|
||||
})
|
||||
assert result['type'] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
|
||||
assert result['data']['username'] == 'test-user'
|
||||
|
||||
|
||||
async def test_saving_loading(data, hass):
|
||||
"""Test saving and loading JSON."""
|
||||
data.add_auth('test-user', 'test-pass')
|
||||
data.add_auth('second-user', 'second-pass')
|
||||
await data.async_save()
|
||||
|
||||
data = hass_auth.Data(hass)
|
||||
await data.async_load()
|
||||
data.validate_login('test-user ', 'test-pass')
|
||||
data.validate_login('second-user ', 'second-pass')
|
||||
|
||||
|
||||
async def test_get_or_create_credentials(hass, data):
|
||||
"""Test that we can get or create credentials."""
|
||||
manager = await auth_manager_from_config(hass, [{
|
||||
'type': 'homeassistant'
|
||||
}], [])
|
||||
provider = manager.auth_providers[0]
|
||||
provider.data = data
|
||||
credentials1 = await provider.async_get_or_create_credentials({
|
||||
'username': 'hello'
|
||||
})
|
||||
with patch.object(provider, 'async_credentials',
|
||||
return_value=mock_coro([credentials1])):
|
||||
credentials2 = await provider.async_get_or_create_credentials({
|
||||
'username': 'hello '
|
||||
})
|
||||
assert credentials1 is credentials2
|
||||
|
||||
|
||||
# Legacy mode
|
||||
|
||||
async def test_legacy_adding_user(legacy_data, hass):
|
||||
"""Test in legacy mode adding a user."""
|
||||
legacy_data.add_auth('test-user', 'test-pass')
|
||||
legacy_data.validate_login('test-user', 'test-pass')
|
||||
|
||||
|
||||
async def test_legacy_adding_user_duplicate_username(legacy_data, hass):
|
||||
"""Test in legacy mode adding a user with duplicate username."""
|
||||
legacy_data.add_auth('test-user', 'test-pass')
|
||||
with pytest.raises(hass_auth.InvalidUser):
|
||||
legacy_data.add_auth('test-user', 'other-pass')
|
||||
|
||||
|
||||
async def test_legacy_validating_password_invalid_password(legacy_data, hass):
|
||||
"""Test in legacy mode validating an invalid password."""
|
||||
legacy_data.add_auth('test-user', 'test-pass')
|
||||
|
||||
with pytest.raises(hass_auth.InvalidAuth):
|
||||
legacy_data.validate_login('test-user', 'invalid-pass')
|
||||
|
||||
|
||||
async def test_legacy_changing_password(legacy_data, hass):
|
||||
"""Test in legacy mode adding a user."""
|
||||
user = 'test-user'
|
||||
legacy_data.add_auth(user, 'test-pass')
|
||||
legacy_data.change_password(user, 'new-pass')
|
||||
|
||||
with pytest.raises(hass_auth.InvalidAuth):
|
||||
legacy_data.validate_login(user, 'test-pass')
|
||||
|
||||
legacy_data.validate_login(user, 'new-pass')
|
||||
|
||||
|
||||
async def test_legacy_changing_password_raises_invalid_user(legacy_data, hass):
|
||||
"""Test in legacy mode that we initialize an empty config."""
|
||||
with pytest.raises(hass_auth.InvalidUser):
|
||||
legacy_data.change_password('non-existing', 'pw')
|
||||
|
||||
|
||||
async def test_legacy_login_flow_validates(legacy_data, hass):
|
||||
"""Test in legacy mode login flow."""
|
||||
legacy_data.add_auth('test-user', 'test-pass')
|
||||
await legacy_data.async_save()
|
||||
|
||||
provider = hass_auth.HassAuthProvider(hass, auth_store.AuthStore(hass),
|
||||
{'type': 'homeassistant'})
|
||||
flow = await provider.async_login_flow({})
|
||||
result = await flow.async_step_init()
|
||||
assert result['type'] == data_entry_flow.RESULT_TYPE_FORM
|
||||
|
||||
result = await flow.async_step_init({
|
||||
'username': 'incorrect-user',
|
||||
'password': 'test-pass',
|
||||
})
|
||||
assert result['type'] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result['errors']['base'] == 'invalid_auth'
|
||||
|
||||
result = await flow.async_step_init({
|
||||
'username': 'test-user',
|
||||
'password': 'incorrect-pass',
|
||||
@ -96,40 +240,43 @@ async def test_login_flow_validates(data, hass):
|
||||
assert result['data']['username'] == 'test-user'
|
||||
|
||||
|
||||
async def test_saving_loading(data, hass):
|
||||
"""Test saving and loading JSON."""
|
||||
data.add_auth('test-user', 'test-pass')
|
||||
data.add_auth('second-user', 'second-pass')
|
||||
await data.async_save()
|
||||
async def test_legacy_saving_loading(legacy_data, hass):
|
||||
"""Test in legacy mode saving and loading JSON."""
|
||||
legacy_data.add_auth('test-user', 'test-pass')
|
||||
legacy_data.add_auth('second-user', 'second-pass')
|
||||
await legacy_data.async_save()
|
||||
|
||||
data = hass_auth.Data(hass)
|
||||
await data.async_load()
|
||||
data.validate_login('test-user', 'test-pass')
|
||||
data.validate_login('second-user', 'second-pass')
|
||||
legacy_data = hass_auth.Data(hass)
|
||||
await legacy_data.async_load()
|
||||
legacy_data.is_legacy = True
|
||||
legacy_data.validate_login('test-user', 'test-pass')
|
||||
legacy_data.validate_login('second-user', 'second-pass')
|
||||
|
||||
with pytest.raises(hass_auth.InvalidAuth):
|
||||
legacy_data.validate_login('test-user ', 'test-pass')
|
||||
|
||||
|
||||
async def test_not_allow_set_id():
|
||||
"""Test we are not allowed to set an ID in config."""
|
||||
hass = Mock()
|
||||
with pytest.raises(vol.Invalid):
|
||||
await auth_provider_from_config(hass, None, {
|
||||
'type': 'homeassistant',
|
||||
'id': 'invalid',
|
||||
})
|
||||
|
||||
|
||||
async def test_new_users_populate_values(hass, data):
|
||||
"""Test that we populate data for new users."""
|
||||
data.add_auth('hello', 'test-pass')
|
||||
await data.async_save()
|
||||
|
||||
async def test_legacy_get_or_create_credentials(hass, legacy_data):
|
||||
"""Test in legacy mode that we can get or create credentials."""
|
||||
manager = await auth_manager_from_config(hass, [{
|
||||
'type': 'homeassistant'
|
||||
}], [])
|
||||
provider = manager.auth_providers[0]
|
||||
credentials = await provider.async_get_or_create_credentials({
|
||||
provider.data = legacy_data
|
||||
credentials1 = await provider.async_get_or_create_credentials({
|
||||
'username': 'hello'
|
||||
})
|
||||
user = await manager.async_get_or_create_user(credentials)
|
||||
assert user.name == 'hello'
|
||||
assert user.is_active
|
||||
|
||||
with patch.object(provider, 'async_credentials',
|
||||
return_value=mock_coro([credentials1])):
|
||||
credentials2 = await provider.async_get_or_create_credentials({
|
||||
'username': 'hello'
|
||||
})
|
||||
assert credentials1 is credentials2
|
||||
|
||||
with patch.object(provider, 'async_credentials',
|
||||
return_value=mock_coro([credentials1])):
|
||||
credentials3 = await provider.async_get_or_create_credentials({
|
||||
'username': 'hello '
|
||||
})
|
||||
assert credentials1 is not credentials3
|
||||
|
||||
@ -168,10 +168,6 @@ async def test_delete_removes_credential(hass, hass_ws_client,
|
||||
client = await hass_ws_client(hass, hass_access_token)
|
||||
|
||||
user = MockUser().add_to_hass(hass)
|
||||
user.credentials.append(
|
||||
await hass.auth.auth_providers[0].async_get_or_create_credentials({
|
||||
'username': 'test-user'}))
|
||||
|
||||
hass_storage[prov_ha.STORAGE_KEY] = {
|
||||
'version': 1,
|
||||
'data': {
|
||||
@ -181,6 +177,10 @@ async def test_delete_removes_credential(hass, hass_ws_client,
|
||||
}
|
||||
}
|
||||
|
||||
user.credentials.append(
|
||||
await hass.auth.auth_providers[0].async_get_or_create_credentials({
|
||||
'username': 'test-user'}))
|
||||
|
||||
await client.send_json({
|
||||
'id': 5,
|
||||
'type': auth_ha.WS_TYPE_DELETE,
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user