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

Allow owner users to change password of any user (#39242)

Browse source
This commit is contained in:
Paulus Schoutsen 2020-08-25 13:49:32 +02:00 • committed by GitHub
parent ef35eea0f6
commit 13df3bce1b
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 251 additions and 127 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

38
homeassistant/auth/providers/homeassistant.py
View file

@ -30,6 +30,15 @@ def _disallow_id(conf: Dict[str, Any]) -> Dict[str, Any]:
CONFIG_SCHEMA = vol.All(AUTH_PROVIDER_SCHEMA, _disallow_id)
async def async_get_provider(hass: HomeAssistant) -> "HassAuthProvider":
"""Get the provider."""
for prv in hass.auth.auth_providers:
if prv.type == "homeassistant":
return cast(HassAuthProvider, prv)
raise RuntimeError("Provider not found")
class InvalidAuth(HomeAssistantError):
"""Raised when we encounter invalid authentication."""
@ -235,6 +244,35 @@ class HassAuthProvider(AuthProvider):
self.data.validate_login, username, password
)
async def async_add_auth(self, username: str, password: str) -> None:
"""Call add_auth on data."""
if self.data is None:
await self.async_initialize()
assert self.data is not None
await self.hass.async_add_executor_job(self.data.add_auth, username, password)
await self.data.async_save()
async def async_remove_auth(self, username: str) -> None:
"""Call remove_auth on data."""
if self.data is None:
await self.async_initialize()
assert self.data is not None
self.data.async_remove_auth(username)
await self.data.async_save()
async def async_change_password(self, username: str, new_password: str) -> None:
"""Call change_password on data."""
if self.data is None:
await self.async_initialize()
assert self.data is not None
await self.hass.async_add_executor_job(
self.data.change_password, username, new_password
)
await self.data.async_save()
async def async_get_or_create_credentials(
self, flow_result: Dict[str, str]
) -> Credentials:

127
homeassistant/components/config/auth_provider_homeassistant.py
View file

@ -3,62 +3,34 @@ import voluptuous as vol
from homeassistant.auth.providers import homeassistant as auth_ha
from homeassistant.components import websocket_api
from homeassistant.components.websocket_api import decorators
from homeassistant.exceptions import Unauthorized
WS_TYPE_CREATE = "config/auth_provider/homeassistant/create"
SCHEMA_WS_CREATE = websocket_api.BASE_COMMAND_MESSAGE_SCHEMA.extend(
async def async_setup(hass):
"""Enable the Home Assistant views."""
hass.components.websocket_api.async_register_command(websocket_create)
hass.components.websocket_api.async_register_command(websocket_delete)
hass.components.websocket_api.async_register_command(websocket_change_password)
hass.components.websocket_api.async_register_command(
websocket_admin_change_password
)
return True
@decorators.websocket_command(
{
vol.Required("type"): WS_TYPE_CREATE,
vol.Required("type"): "config/auth_provider/homeassistant/create",
vol.Required("user_id"): str,
vol.Required("username"): str,
vol.Required("password"): str,
}
)
WS_TYPE_DELETE = "config/auth_provider/homeassistant/delete"
SCHEMA_WS_DELETE = websocket_api.BASE_COMMAND_MESSAGE_SCHEMA.extend(
{vol.Required("type"): WS_TYPE_DELETE, vol.Required("username"): str}
)
WS_TYPE_CHANGE_PASSWORD = "config/auth_provider/homeassistant/change_password"
SCHEMA_WS_CHANGE_PASSWORD = websocket_api.BASE_COMMAND_MESSAGE_SCHEMA.extend(
{
vol.Required("type"): WS_TYPE_CHANGE_PASSWORD,
vol.Required("current_password"): str,
vol.Required("new_password"): str,
}
)
async def async_setup(hass):
"""Enable the Home Assistant views."""
hass.components.websocket_api.async_register_command(
WS_TYPE_CREATE, websocket_create, SCHEMA_WS_CREATE
)
hass.components.websocket_api.async_register_command(
WS_TYPE_DELETE, websocket_delete, SCHEMA_WS_DELETE
)
hass.components.websocket_api.async_register_command(
WS_TYPE_CHANGE_PASSWORD, websocket_change_password, SCHEMA_WS_CHANGE_PASSWORD
)
return True
def _get_provider(hass):
"""Get homeassistant auth provider."""
for prv in hass.auth.auth_providers:
if prv.type == "homeassistant":
return prv
raise RuntimeError("Provider not found")
@websocket_api.require_admin
@websocket_api.async_response
async def websocket_create(hass, connection, msg):
"""Create credentials and attach to a user."""
provider = _get_provider(hass)
await provider.async_initialize()
provider = await auth_ha.async_get_provider(hass)
user = await hass.auth.async_get_user(msg["user_id"])
if user is None:
@ -78,9 +50,7 @@ async def websocket_create(hass, connection, msg):
return
try:
await hass.async_add_executor_job(
provider.data.add_auth, msg["username"], msg["password"]
)
await provider.async_add_auth(msg["username"], msg["password"])
except auth_ha.InvalidUser:
connection.send_message(
websocket_api.error_message(
@ -94,17 +64,20 @@ async def websocket_create(hass, connection, msg):
)
await hass.auth.async_link_user(user, credentials)
await provider.data.async_save()
connection.send_message(websocket_api.result_message(msg["id"]))
@decorators.websocket_command(
{
vol.Required("type"): "config/auth_provider/homeassistant/delete",
vol.Required("username"): str,
}
)
@websocket_api.require_admin
@websocket_api.async_response
async def websocket_delete(hass, connection, msg):
"""Delete username and related credential."""
provider = _get_provider(hass)
await provider.async_initialize()
provider = await auth_ha.async_get_provider(hass)
credentials = await provider.async_get_or_create_credentials(
{"username": msg["username"]}
)
@ -118,8 +91,7 @@ async def websocket_delete(hass, connection, msg):
return
try:
provider.data.async_remove_auth(msg["username"])
await provider.data.async_save()
await provider.async_remove_auth(msg["username"])
except auth_ha.InvalidUser:
connection.send_message(
websocket_api.error_message(
@ -131,9 +103,16 @@ async def websocket_delete(hass, connection, msg):
connection.send_message(websocket_api.result_message(msg["id"]))
@decorators.websocket_command(
{
vol.Required("type"): "config/auth_provider/homeassistant/change_password",
vol.Required("current_password"): str,
vol.Required("new_password"): str,
}
)
@websocket_api.async_response
async def websocket_change_password(hass, connection, msg):
"""Change user password."""
"""Change current user password."""
user = connection.user
if user is None:
connection.send_message(
@ -141,9 +120,7 @@ async def websocket_change_password(hass, connection, msg):
)
return
provider = _get_provider(hass)
await provider.async_initialize()
provider = await auth_ha.async_get_provider(hass)
username = None
for credential in user.credentials:
if credential.auth_provider_type == provider.type:
@ -168,9 +145,35 @@ async def websocket_change_password(hass, connection, msg):
)
return
await hass.async_add_executor_job(
provider.data.change_password, username, msg["new_password"]
)
await provider.data.async_save()
await provider.async_change_password(username, msg["new_password"])
connection.send_message(websocket_api.result_message(msg["id"]))
@decorators.websocket_command(
{
vol.Required(
"type"
): "config/auth_provider/homeassistant/admin_change_password",
vol.Required("username"): str,
vol.Required("password"): str,
}
)
@decorators.require_admin
@decorators.async_response
async def websocket_admin_change_password(hass, connection, msg):
"""Change password of any user."""
if not connection.user.is_owner:
raise Unauthorized(context=connection.context(msg))
provider = await auth_ha.async_get_provider(hass)
try:
await provider.async_change_password(msg["username"], msg["password"])
connection.send_message(websocket_api.result_message(msg["id"]))
except auth_ha.InvalidUser:
connection.send_message(
websocket_api.error_message(
msg["id"], "credentials_not_found", "Credentials not found"
)
)
return

213
tests/components/config/test_auth_provider_homeassistant.py
View file

@ -4,7 +4,7 @@ import pytest
from homeassistant.auth.providers import homeassistant as prov_ha
from homeassistant.components.config import auth_provider_homeassistant as auth_ha
from tests.common import MockUser, register_auth_provider
from tests.common import CLIENT_ID, MockUser, register_auth_provider
@pytest.fixture(autouse=True)
@ -16,17 +16,44 @@ def setup_config(hass):
hass.loop.run_until_complete(auth_ha.async_setup(hass))
async def test_create_auth_system_generated_user(
hass, hass_access_token, hass_ws_client
):
@pytest.fixture
async def auth_provider(hass):
"""Hass auth provider."""
provider = hass.auth.auth_providers[0]
await provider.async_initialize()
return provider
@pytest.fixture
async def owner_access_token(hass, hass_owner_user):
"""Access token for owner user."""
refresh_token = await hass.auth.async_create_refresh_token(
hass_owner_user, CLIENT_ID
)
return hass.auth.async_create_access_token(refresh_token)
@pytest.fixture
async def test_user_credential(hass, auth_provider):
"""Add a test user."""
await hass.async_add_executor_job(
auth_provider.data.add_auth, "test-user", "test-pass"
)
return await auth_provider.async_get_or_create_credentials(
{"username": "test-user"}
)
async def test_create_auth_system_generated_user(hass, hass_ws_client):
"""Test we can't add auth to system generated users."""
system_user = MockUser(system_generated=True).add_to_hass(hass)
client = await hass_ws_client(hass, hass_access_token)
client = await hass_ws_client(hass)
await client.send_json(
{
"id": 5,
"type": auth_ha.WS_TYPE_CREATE,
"type": "config/auth_provider/homeassistant/create",
"user_id": system_user.id,
"username": "test-user",
"password": "test-pass",
@ -44,14 +71,14 @@ async def test_create_auth_user_already_credentials():
# assert False
async def test_create_auth_unknown_user(hass_ws_client, hass, hass_access_token):
async def test_create_auth_unknown_user(hass_ws_client, hass):
"""Test create pointing at unknown user."""
client = await hass_ws_client(hass, hass_access_token)
client = await hass_ws_client(hass)
await client.send_json(
{
"id": 5,
"type": auth_ha.WS_TYPE_CREATE,
"type": "config/auth_provider/homeassistant/create",
"user_id": "test-id",
"username": "test-user",
"password": "test-pass",
@ -73,7 +100,7 @@ async def test_create_auth_requires_admin(
await client.send_json(
{
"id": 5,
"type": auth_ha.WS_TYPE_CREATE,
"type": "config/auth_provider/homeassistant/create",
"user_id": "test-id",
"username": "test-user",
"password": "test-pass",
@ -85,9 +112,9 @@ async def test_create_auth_requires_admin(
assert result["error"]["code"] == "unauthorized"
async def test_create_auth(hass, hass_ws_client, hass_access_token, hass_storage):
async def test_create_auth(hass, hass_ws_client, hass_storage):
"""Test create auth command works."""
client = await hass_ws_client(hass, hass_access_token)
client = await hass_ws_client(hass)
user = MockUser().add_to_hass(hass)
assert len(user.credentials) == 0
@ -95,7 +122,7 @@ async def test_create_auth(hass, hass_ws_client, hass_access_token, hass_storage
await client.send_json(
{
"id": 5,
"type": auth_ha.WS_TYPE_CREATE,
"type": "config/auth_provider/homeassistant/create",
"user_id": user.id,
"username": "test-user",
"password": "test-pass",
@ -114,11 +141,9 @@ async def test_create_auth(hass, hass_ws_client, hass_access_token, hass_storage
assert entry["username"] == "test-user"
async def test_create_auth_duplicate_username(
hass, hass_ws_client, hass_access_token, hass_storage
):
async def test_create_auth_duplicate_username(hass, hass_ws_client, hass_storage):
"""Test we can't create auth with a duplicate username."""
client = await hass_ws_client(hass, hass_access_token)
client = await hass_ws_client(hass)
user = MockUser().add_to_hass(hass)
hass_storage[prov_ha.STORAGE_KEY] = {
@ -129,7 +154,7 @@ async def test_create_auth_duplicate_username(
await client.send_json(
{
"id": 5,
"type": auth_ha.WS_TYPE_CREATE,
"type": "config/auth_provider/homeassistant/create",
"user_id": user.id,
"username": "test-user",
"password": "test-pass",
@ -141,11 +166,9 @@ async def test_create_auth_duplicate_username(
assert result["error"]["code"] == "username_exists"
async def test_delete_removes_just_auth(
hass_ws_client, hass, hass_storage, hass_access_token
):
async def test_delete_removes_just_auth(hass_ws_client, hass, hass_storage):
"""Test deleting an auth without being connected to a user."""
client = await hass_ws_client(hass, hass_access_token)
client = await hass_ws_client(hass)
hass_storage[prov_ha.STORAGE_KEY] = {
"version": 1,
@ -153,7 +176,11 @@ async def test_delete_removes_just_auth(
}
await client.send_json(
{"id": 5, "type": auth_ha.WS_TYPE_DELETE, "username": "test-user"}
{
"id": 5,
"type": "config/auth_provider/homeassistant/delete",
"username": "test-user",
}
)
result = await client.receive_json()
@ -161,11 +188,9 @@ async def test_delete_removes_just_auth(
assert len(hass_storage[prov_ha.STORAGE_KEY]["data"]["users"]) == 0
async def test_delete_removes_credential(
hass, hass_ws_client, hass_access_token, hass_storage
):
async def test_delete_removes_credential(hass, hass_ws_client, hass_storage):
"""Test deleting auth that is connected to a user."""
client = await hass_ws_client(hass, hass_access_token)
client = await hass_ws_client(hass)
user = MockUser().add_to_hass(hass)
hass_storage[prov_ha.STORAGE_KEY] = {
@ -180,7 +205,11 @@ async def test_delete_removes_credential(
)
await client.send_json(
{"id": 5, "type": auth_ha.WS_TYPE_DELETE, "username": "test-user"}
{
"id": 5,
"type": "config/auth_provider/homeassistant/delete",
"username": "test-user",
}
)
result = await client.receive_json()
@ -193,7 +222,11 @@ async def test_delete_requires_admin(hass, hass_ws_client, hass_read_only_access
client = await hass_ws_client(hass, hass_read_only_access_token)
await client.send_json(
{"id": 5, "type": auth_ha.WS_TYPE_DELETE, "username": "test-user"}
{
"id": 5,
"type": "config/auth_provider/homeassistant/delete",
"username": "test-user",
}
)
result = await client.receive_json()
@ -201,12 +234,16 @@ async def test_delete_requires_admin(hass, hass_ws_client, hass_read_only_access
assert result["error"]["code"] == "unauthorized"
async def test_delete_unknown_auth(hass, hass_ws_client, hass_access_token):
async def test_delete_unknown_auth(hass, hass_ws_client):
"""Test trying to delete an unknown auth username."""
client = await hass_ws_client(hass, hass_access_token)
client = await hass_ws_client(hass)
await client.send_json(
{"id": 5, "type": auth_ha.WS_TYPE_DELETE, "username": "test-user"}
{
"id": 5,
"type": "config/auth_provider/homeassistant/delete",
"username": "test-user",
}
)
result = await client.receive_json()
@ -214,25 +251,17 @@ async def test_delete_unknown_auth(hass, hass_ws_client, hass_access_token):
assert result["error"]["code"] == "auth_not_found"
async def test_change_password(hass, hass_ws_client, hass_access_token):
async def test_change_password(
hass, hass_ws_client, hass_admin_user, auth_provider, test_user_credential
):
"""Test that change password succeeds with valid password."""
provider = hass.auth.auth_providers[0]
await provider.async_initialize()
await hass.async_add_executor_job(provider.data.add_auth, "test-user", "test-pass")
await hass.auth.async_link_user(hass_admin_user, test_user_credential)
credentials = await provider.async_get_or_create_credentials(
{"username": "test-user"}
)
refresh_token = await hass.auth.async_validate_access_token(hass_access_token)
user = refresh_token.user
await hass.auth.async_link_user(user, credentials)
client = await hass_ws_client(hass, hass_access_token)
client = await hass_ws_client(hass)
await client.send_json(
{
"id": 6,
"type": auth_ha.WS_TYPE_CHANGE_PASSWORD,
"type": "config/auth_provider/homeassistant/change_password",
"current_password": "test-pass",
"new_password": "new-pass",
}
@ -240,28 +269,20 @@ async def test_change_password(hass, hass_ws_client, hass_access_token):
result = await client.receive_json()
assert result["success"], result
await provider.async_validate_login("test-user", "new-pass")
await auth_provider.async_validate_login("test-user", "new-pass")
async def test_change_password_wrong_pw(hass, hass_ws_client, hass_access_token):
async def test_change_password_wrong_pw(
hass, hass_ws_client, hass_admin_user, auth_provider, test_user_credential
):
"""Test that change password fails with invalid password."""
provider = hass.auth.auth_providers[0]
await provider.async_initialize()
await hass.async_add_executor_job(provider.data.add_auth, "test-user", "test-pass")
await hass.auth.async_link_user(hass_admin_user, test_user_credential)
credentials = await provider.async_get_or_create_credentials(
{"username": "test-user"}
)
refresh_token = await hass.auth.async_validate_access_token(hass_access_token)
user = refresh_token.user
await hass.auth.async_link_user(user, credentials)
client = await hass_ws_client(hass, hass_access_token)
client = await hass_ws_client(hass)
await client.send_json(
{
"id": 6,
"type": auth_ha.WS_TYPE_CHANGE_PASSWORD,
"type": "config/auth_provider/homeassistant/change_password",
"current_password": "wrong-pass",
"new_password": "new-pass",
}
@ -271,17 +292,17 @@ async def test_change_password_wrong_pw(hass, hass_ws_client, hass_access_token)
assert not result["success"], result
assert result["error"]["code"] == "invalid_password"
with pytest.raises(prov_ha.InvalidAuth):
await provider.async_validate_login("test-user", "new-pass")
await auth_provider.async_validate_login("test-user", "new-pass")
async def test_change_password_no_creds(hass, hass_ws_client, hass_access_token):
async def test_change_password_no_creds(hass, hass_ws_client):
"""Test that change password fails with no credentials."""
client = await hass_ws_client(hass, hass_access_token)
client = await hass_ws_client(hass)
await client.send_json(
{
"id": 6,
"type": auth_ha.WS_TYPE_CHANGE_PASSWORD,
"type": "config/auth_provider/homeassistant/change_password",
"current_password": "test-pass",
"new_password": "new-pass",
}
@ -290,3 +311,65 @@ async def test_change_password_no_creds(hass, hass_ws_client, hass_access_token)
result = await client.receive_json()
assert not result["success"], result
assert result["error"]["code"] == "credentials_not_found"
async def test_admin_change_password_not_owner(
hass, hass_ws_client, auth_provider, test_user_credential
):
"""Test that change password fails when not owner."""
client = await hass_ws_client(hass)
await client.send_json(
{
"id": 6,
"type": "config/auth_provider/homeassistant/admin_change_password",
"username": "test-user",
"password": "new-pass",
}
)
result = await client.receive_json()
assert not result["success"], result
assert result["error"]["code"] == "unauthorized"
# Validate old login still works
await auth_provider.async_validate_login("test-user", "test-pass")
async def test_admin_change_password_no_creds(hass, hass_ws_client, owner_access_token):
"""Test that change password fails with unknown credentials."""
client = await hass_ws_client(hass, owner_access_token)
await client.send_json(
{
"id": 6,
"type": "config/auth_provider/homeassistant/admin_change_password",
"username": "non-existing",
"password": "new-pass",
}
)
result = await client.receive_json()
assert not result["success"], result
assert result["error"]["code"] == "credentials_not_found"
async def test_admin_change_password(
hass, hass_ws_client, owner_access_token, auth_provider, test_user_credential
):
"""Test that owners can change any password."""
client = await hass_ws_client(hass, owner_access_token)
await client.send_json(
{
"id": 6,
"type": "config/auth_provider/homeassistant/admin_change_password",
"username": "test-user",
"password": "new-pass",
}
)
result = await client.receive_json()
assert result["success"], result
await auth_provider.async_validate_login("test-user", "new-pass")