hass-core/tests/auth/test_init.py

Ignoring revisions in .git-blame-ignore-revs. Click here to bypass and see the normal blame view.

1381 lines
48 KiB
Python
Raw Normal View History

"""Tests for the Home Assistant auth module."""
from datetime import timedelta
import time
2023-02-20 11:42:56 +01:00
from typing import Any
from unittest.mock import patch
2023-03-22 16:03:41 -10:00
from freezegun import freeze_time
import jwt
import pytest
import voluptuous as vol
from homeassistant import auth, data_entry_flow
from homeassistant.auth import (
2022-06-09 17:49:02 -10:00
EVENT_USER_UPDATED,
InvalidAuthError,
auth_store,
const as auth_const,
models as auth_models,
)
2021-11-29 14:01:03 -08:00
from homeassistant.auth.const import GROUP_ID_ADMIN, MFA_SESSION_EXPIRATION
2023-02-20 11:42:56 +01:00
from homeassistant.auth.models import Credentials
from homeassistant.core import HomeAssistant, callback
from homeassistant.util import dt as dt_util
from tests.common import (
CLIENT_ID,
MockUser,
async_capture_events,
async_fire_time_changed,
ensure_auth_manager_loaded,
flush_store,
)
@pytest.fixture
def mock_hass(hass: HomeAssistant) -> HomeAssistant:
"""Home Assistant mock with minimum amount of data set to make it work with auth."""
return hass
2023-02-20 11:42:56 +01:00
async def test_auth_manager_from_config_validates_config(mock_hass) -> None:
"""Test get auth providers."""
with pytest.raises(vol.Invalid):
manager = await auth.auth_manager_from_config(
mock_hass,
[
{"name": "Test Name", "type": "insecure_example", "users": []},
{
2020-02-13 17:27:00 +01:00
"name": "Invalid configuration because no users",
"type": "insecure_example",
"id": "invalid_config",
},
2019-07-31 12:25:30 -07:00
],
[],
)
2019-07-31 12:25:30 -07:00
manager = await auth.auth_manager_from_config(
mock_hass,
[
{"name": "Test Name", "type": "insecure_example", "users": []},
2019-07-31 12:25:30 -07:00
{
"name": "Test Name 2",
"type": "insecure_example",
"id": "another",
"users": [],
},
2019-07-31 12:25:30 -07:00
],
[],
)
2019-07-31 12:25:30 -07:00
providers = [
{"name": provider.name, "id": provider.id, "type": provider.type}
for provider in manager.auth_providers
]
2019-07-31 12:25:30 -07:00
assert providers == [
{"name": "Test Name", "type": "insecure_example", "id": None},
{"name": "Test Name 2", "type": "insecure_example", "id": "another"},
]
2023-02-20 11:42:56 +01:00
async def test_auth_manager_from_config_auth_modules(mock_hass) -> None:
"""Test get auth modules."""
with pytest.raises(vol.Invalid):
manager = await auth.auth_manager_from_config(
mock_hass,
[
{"name": "Test Name", "type": "insecure_example", "users": []},
{
"name": "Test Name 2",
"type": "insecure_example",
"id": "another",
"users": [],
},
],
[
{"name": "Module 1", "type": "insecure_example", "data": []},
{
2020-02-13 17:27:00 +01:00
"name": "Invalid configuration because no data",
"type": "insecure_example",
"id": "another",
},
],
2019-07-31 12:25:30 -07:00
)
manager = await auth.auth_manager_from_config(
mock_hass,
[
{"name": "Test Name", "type": "insecure_example", "users": []},
2019-07-31 12:25:30 -07:00
{
"name": "Test Name 2",
"type": "insecure_example",
"id": "another",
"users": [],
},
],
[
{"name": "Module 1", "type": "insecure_example", "data": []},
{
"name": "Module 2",
"type": "insecure_example",
"id": "another",
"data": [],
},
2019-07-31 12:25:30 -07:00
],
)
providers = [
{"name": provider.name, "type": provider.type, "id": provider.id}
for provider in manager.auth_providers
]
assert providers == [
{"name": "Test Name", "type": "insecure_example", "id": None},
{"name": "Test Name 2", "type": "insecure_example", "id": "another"},
]
2019-07-31 12:25:30 -07:00
modules = [
{"name": module.name, "type": module.type, "id": module.id}
for module in manager.auth_mfa_modules
]
assert modules == [
{"name": "Module 1", "type": "insecure_example", "id": "insecure_example"},
{"name": "Module 2", "type": "insecure_example", "id": "another"},
]
async def test_create_new_user(hass: HomeAssistant) -> None:
"""Test creating new user."""
2018-10-11 17:06:51 +02:00
events = []
@callback
def user_added(event):
events.append(event)
hass.bus.async_listen("user_added", user_added)
2019-07-31 12:25:30 -07:00
manager = await auth.auth_manager_from_config(
hass,
[
{
"type": "insecure_example",
"users": [
{
"username": "test-user",
"password": "test-pass",
"name": "Test Name",
}
2019-07-31 12:25:30 -07:00
],
}
],
[],
)
2019-07-31 12:25:30 -07:00
step = await manager.login_flow.async_init(("insecure_example", None))
assert step["type"] == data_entry_flow.FlowResultType.FORM
2019-07-31 12:25:30 -07:00
step = await manager.login_flow.async_configure(
step["flow_id"], {"username": "test-user", "password": "test-pass"}
)
assert step["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
credential = step["result"]
assert credential is not None
user = await manager.async_get_or_create_user(credential)
assert user is not None
assert user.is_owner is False
assert user.name == "Test Name"
2018-10-11 17:06:51 +02:00
await hass.async_block_till_done()
assert len(events) == 1
assert events[0].data["user_id"] == user.id
2023-02-20 11:42:56 +01:00
async def test_login_as_existing_user(mock_hass) -> None:
"""Test login as existing user."""
manager = await auth.auth_manager_from_config(
mock_hass,
[
{
"type": "insecure_example",
"users": [
{
"username": "test-user",
"password": "test-pass",
"name": "Test Name",
2019-07-31 12:25:30 -07:00
}
],
}
2019-07-31 12:25:30 -07:00
],
[],
)
mock_hass.auth = manager
ensure_auth_manager_loaded(manager)
2018-07-10 20:33:03 +02:00
# Add a fake user that we're not going to log in with
user = MockUser(
id="mock-user2", is_owner=False, is_active=False, name="Not user"
).add_to_auth_manager(manager)
2018-07-13 11:43:08 +02:00
user.credentials.append(
auth_models.Credentials(
2018-07-10 20:33:03 +02:00
id="mock-id2",
auth_provider_type="insecure_example",
auth_provider_id=None,
data={"username": "other-user"},
is_new=False,
2019-07-31 12:25:30 -07:00
)
2018-07-10 20:33:03 +02:00
)
# Add fake user with credentials for example auth provider.
user = MockUser(
id="mock-user", is_owner=False, is_active=False, name="Paulus"
).add_to_auth_manager(manager)
2018-07-13 11:43:08 +02:00
user.credentials.append(
auth_models.Credentials(
id="mock-id",
auth_provider_type="insecure_example",
auth_provider_id=None,
data={"username": "test-user"},
is_new=False,
)
2019-07-31 12:25:30 -07:00
)
step = await manager.login_flow.async_init(("insecure_example", None))
assert step["type"] == data_entry_flow.FlowResultType.FORM
2019-07-31 12:25:30 -07:00
step = await manager.login_flow.async_configure(
step["flow_id"], {"username": "test-user", "password": "test-pass"}
)
assert step["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
2019-07-31 12:25:30 -07:00
credential = step["result"]
user = await manager.async_get_user_by_credentials(credential)
assert user is not None
assert user.id == "mock-user"
assert user.is_owner is False
assert user.is_active is False
assert user.name == "Paulus"
2023-02-20 11:42:56 +01:00
async def test_linking_user_to_two_auth_providers(
hass: HomeAssistant, hass_storage: dict[str, Any]
) -> None:
"""Test linking user to two auth providers."""
manager = await auth.auth_manager_from_config(
hass,
[
{
"type": "insecure_example",
"users": [{"username": "test-user", "password": "test-pass"}],
},
{
"type": "insecure_example",
"id": "another-provider",
"users": [{"username": "another-user", "password": "another-password"}],
},
2019-07-31 12:25:30 -07:00
],
[],
)
2019-07-31 12:25:30 -07:00
step = await manager.login_flow.async_init(("insecure_example", None))
step = await manager.login_flow.async_configure(
step["flow_id"], {"username": "test-user", "password": "test-pass"}
)
credential = step["result"]
user = await manager.async_get_or_create_user(credential)
assert user is not None
step = await manager.login_flow.async_init(
("insecure_example", "another-provider"), context={"credential_only": True}
)
step = await manager.login_flow.async_configure(
step["flow_id"], {"username": "another-user", "password": "another-password"}
)
new_credential = step["result"]
await manager.async_link_user(user, new_credential)
assert len(user.credentials) == 2
# Linking it again to same user is a no-op
await manager.async_link_user(user, new_credential)
assert len(user.credentials) == 2
# Linking a credential to a user while the credential is already linked to another user should raise
user_2 = await manager.async_create_user("User 2")
with pytest.raises(ValueError):
await manager.async_link_user(user_2, new_credential)
assert len(user_2.credentials) == 0
2023-02-20 11:42:56 +01:00
async def test_saving_loading(
hass: HomeAssistant, hass_storage: dict[str, Any]
) -> None:
"""Test storing and saving data.
Creates one of each type that we store to test we restore correctly.
"""
manager = await auth.auth_manager_from_config(
hass,
[
{
"type": "insecure_example",
"users": [{"username": "test-user", "password": "test-pass"}],
}
2019-07-31 12:25:30 -07:00
],
[],
)
2019-07-31 12:25:30 -07:00
step = await manager.login_flow.async_init(("insecure_example", None))
step = await manager.login_flow.async_configure(
step["flow_id"], {"username": "test-user", "password": "test-pass"}
)
credential = step["result"]
user = await manager.async_get_or_create_user(credential)
await manager.async_activate_user(user)
# the first refresh token will be used to create access token
refresh_token = await manager.async_create_refresh_token(
user, CLIENT_ID, credential=credential
)
manager.async_create_access_token(refresh_token, "192.168.0.1")
# the second refresh token will not be used
await manager.async_create_refresh_token(
user, "dummy-client", credential=credential
)
await flush_store(manager._store._store)
2018-07-13 11:43:08 +02:00
store2 = auth_store.AuthStore(hass)
await store2.async_load()
users = await store2.async_get_users()
assert len(users) == 1
assert users[0].permissions == user.permissions
assert users[0] == user
assert len(users[0].refresh_tokens) == 2
for r_token in users[0].refresh_tokens.values():
if r_token.client_id == CLIENT_ID:
# verify the first refresh token
assert r_token.last_used_at is not None
assert r_token.last_used_ip == "192.168.0.1"
elif r_token.client_id == "dummy-client":
# verify the second refresh token
assert r_token.last_used_at is None
assert r_token.last_used_ip is None
else:
2023-01-27 11:10:29 +01:00
pytest.fail(f"Unknown client_id: {r_token.client_id}")
async def test_cannot_retrieve_expired_access_token(hass: HomeAssistant) -> None:
"""Test that we cannot retrieve expired access tokens."""
manager = await auth.auth_manager_from_config(hass, [], [])
user = MockUser().add_to_auth_manager(manager)
refresh_token = await manager.async_create_refresh_token(user, CLIENT_ID)
assert refresh_token.user.id is user.id
assert refresh_token.client_id == CLIENT_ID
access_token = manager.async_create_access_token(refresh_token)
assert manager.async_validate_access_token(access_token) is refresh_token
2019-07-31 12:25:30 -07:00
# We patch time directly here because we want the access token to be created with
# an expired time, but we do not want to freeze time so that jwt will compare it
# to the patched time. If we freeze time for the test it will be frozen for jwt
# as well and the token will not be expired.
2018-07-13 11:43:08 +02:00
with patch(
"homeassistant.auth.time.time",
return_value=time.time()
- auth_const.ACCESS_TOKEN_EXPIRATION.total_seconds()
- 11,
):
access_token = manager.async_create_access_token(refresh_token)
assert manager.async_validate_access_token(access_token) is None
async def test_generating_system_user(hass: HomeAssistant) -> None:
"""Test that we can add a system user."""
2018-10-11 17:06:51 +02:00
events = []
@callback
def user_added(event):
events.append(event)
hass.bus.async_listen("user_added", user_added)
manager = await auth.auth_manager_from_config(hass, [], [])
user = await manager.async_create_system_user("Hass.io")
token = await manager.async_create_refresh_token(user)
assert user.system_generated
2021-11-29 14:01:03 -08:00
assert user.groups == []
assert not user.local_only
assert token is not None
assert token.client_id is None
assert token.token_type == auth.models.TOKEN_TYPE_SYSTEM
assert token.expire_at is None
2018-10-11 17:06:51 +02:00
await hass.async_block_till_done()
assert len(events) == 1
assert events[0].data["user_id"] == user.id
2021-11-29 14:01:03 -08:00
# Passing arguments
user = await manager.async_create_system_user(
"Hass.io", group_ids=[GROUP_ID_ADMIN], local_only=True
)
token = await manager.async_create_refresh_token(user)
assert user.system_generated
assert user.is_admin
assert user.local_only
assert token is not None
assert token.client_id is None
assert token.token_type == auth.models.TOKEN_TYPE_SYSTEM
assert token.expire_at is None
2021-11-29 14:01:03 -08:00
await hass.async_block_till_done()
assert len(events) == 2
assert events[1].data["user_id"] == user.id
async def test_refresh_token_requires_client_for_user(hass: HomeAssistant) -> None:
"""Test create refresh token for a user with client_id."""
manager = await auth.auth_manager_from_config(hass, [], [])
user = MockUser().add_to_auth_manager(manager)
assert user.system_generated is False
with pytest.raises(ValueError):
await manager.async_create_refresh_token(user)
token = await manager.async_create_refresh_token(user, CLIENT_ID)
assert token is not None
assert token.client_id == CLIENT_ID
assert token.token_type == auth_models.TOKEN_TYPE_NORMAL
# default access token expiration
assert token.access_token_expiration == auth_const.ACCESS_TOKEN_EXPIRATION
async def test_refresh_token_not_requires_client_for_system_user(
hass: HomeAssistant,
) -> None:
"""Test create refresh token for a system user w/o client_id."""
manager = await auth.auth_manager_from_config(hass, [], [])
user = await manager.async_create_system_user("Hass.io")
assert user.system_generated is True
with pytest.raises(ValueError):
await manager.async_create_refresh_token(user, CLIENT_ID)
token = await manager.async_create_refresh_token(user)
assert token is not None
assert token.client_id is None
assert token.token_type == auth_models.TOKEN_TYPE_SYSTEM
async def test_refresh_token_with_specific_access_token_expiration(
hass: HomeAssistant,
) -> None:
"""Test create a refresh token with specific access token expiration."""
manager = await auth.auth_manager_from_config(hass, [], [])
user = MockUser().add_to_auth_manager(manager)
token = await manager.async_create_refresh_token(
user, CLIENT_ID, access_token_expiration=timedelta(days=100)
2019-07-31 12:25:30 -07:00
)
assert token is not None
assert token.client_id == CLIENT_ID
assert token.access_token_expiration == timedelta(days=100)
assert token.token_type == auth.models.TOKEN_TYPE_NORMAL
assert token.expire_at is not None
async def test_refresh_token_type(hass: HomeAssistant) -> None:
"""Test create a refresh token with token type."""
manager = await auth.auth_manager_from_config(hass, [], [])
user = MockUser().add_to_auth_manager(manager)
with pytest.raises(ValueError):
await manager.async_create_refresh_token(
user, CLIENT_ID, token_type=auth_models.TOKEN_TYPE_SYSTEM
)
token = await manager.async_create_refresh_token(
user, CLIENT_ID, token_type=auth_models.TOKEN_TYPE_NORMAL
)
assert token is not None
assert token.client_id == CLIENT_ID
assert token.token_type == auth_models.TOKEN_TYPE_NORMAL
async def test_refresh_token_type_long_lived_access_token(hass: HomeAssistant) -> None:
"""Test create a refresh token has long-lived access token type."""
manager = await auth.auth_manager_from_config(hass, [], [])
user = MockUser().add_to_auth_manager(manager)
with pytest.raises(ValueError):
await manager.async_create_refresh_token(
user, token_type=auth_models.TOKEN_TYPE_LONG_LIVED_ACCESS_TOKEN
)
token = await manager.async_create_refresh_token(
user,
client_name="GPS LOGGER",
client_icon="mdi:home",
token_type=auth_models.TOKEN_TYPE_LONG_LIVED_ACCESS_TOKEN,
)
assert token is not None
assert token.client_id is None
assert token.client_name == "GPS LOGGER"
assert token.client_icon == "mdi:home"
assert token.token_type == auth_models.TOKEN_TYPE_LONG_LIVED_ACCESS_TOKEN
assert token.expire_at is None
2023-02-20 11:42:56 +01:00
async def test_refresh_token_provider_validation(mock_hass) -> None:
"""Test that creating access token from refresh token checks with provider."""
manager = await auth.auth_manager_from_config(
mock_hass,
[
{
"type": "insecure_example",
"users": [{"username": "test-user", "password": "test-pass"}],
}
],
[],
)
credential = auth_models.Credentials(
id="mock-credential-id",
auth_provider_type="insecure_example",
auth_provider_id=None,
data={"username": "test-user"},
is_new=False,
)
user = MockUser().add_to_auth_manager(manager)
user.credentials.append(credential)
refresh_token = await manager.async_create_refresh_token(
user, CLIENT_ID, credential=credential
)
ip = "127.0.0.1"
assert manager.async_create_access_token(refresh_token, ip) is not None
with (
patch(
"homeassistant.auth.providers.insecure_example.ExampleAuthProvider.async_validate_refresh_token",
side_effect=InvalidAuthError("Invalid access"),
) as call,
pytest.raises(InvalidAuthError),
):
manager.async_create_access_token(refresh_token, ip)
call.assert_called_with(refresh_token, ip)
2023-02-20 11:42:56 +01:00
async def test_cannot_deactive_owner(mock_hass) -> None:
"""Test that we cannot deactivate the owner."""
manager = await auth.auth_manager_from_config(mock_hass, [], [])
owner = MockUser(is_owner=True).add_to_auth_manager(manager)
with pytest.raises(ValueError):
await manager.async_deactivate_user(owner)
async def test_remove_refresh_token(hass: HomeAssistant) -> None:
"""Test that we can remove a refresh token."""
manager = await auth.auth_manager_from_config(hass, [], [])
user = MockUser().add_to_auth_manager(manager)
refresh_token = await manager.async_create_refresh_token(user, CLIENT_ID)
access_token = manager.async_create_access_token(refresh_token)
manager.async_remove_refresh_token(refresh_token)
assert manager.async_get_refresh_token(refresh_token.id) is None
assert manager.async_validate_access_token(access_token) is None
async def test_remove_expired_refresh_token(hass: HomeAssistant) -> None:
"""Test that expired refresh tokens are deleted."""
manager = await auth.auth_manager_from_config(hass, [], [])
user = MockUser().add_to_auth_manager(manager)
now = dt_util.utcnow()
with freeze_time(now):
refresh_token1 = await manager.async_create_refresh_token(user, CLIENT_ID)
assert (
refresh_token1.expire_at
== now.timestamp() + timedelta(days=90).total_seconds()
)
with freeze_time(now + timedelta(days=30)):
async_fire_time_changed(hass, now + timedelta(days=30))
refresh_token2 = await manager.async_create_refresh_token(user, CLIENT_ID)
assert (
refresh_token2.expire_at
== now.timestamp() + timedelta(days=120).total_seconds()
)
with freeze_time(now + timedelta(days=89, hours=23)):
async_fire_time_changed(hass, now + timedelta(days=89, hours=23))
await hass.async_block_till_done()
assert manager.async_get_refresh_token(refresh_token1.id)
assert manager.async_get_refresh_token(refresh_token2.id)
with freeze_time(now + timedelta(days=90, seconds=5)):
async_fire_time_changed(hass, now + timedelta(days=90, seconds=5))
await hass.async_block_till_done()
assert manager.async_get_refresh_token(refresh_token1.id) is None
assert manager.async_get_refresh_token(refresh_token2.id)
with freeze_time(now + timedelta(days=120, seconds=5)):
async_fire_time_changed(hass, now + timedelta(days=120, seconds=5))
await hass.async_block_till_done()
assert manager.async_get_refresh_token(refresh_token1.id) is None
assert manager.async_get_refresh_token(refresh_token2.id) is None
async def test_update_expire_at_refresh_token(hass: HomeAssistant) -> None:
"""Test that expire at is updated when refresh token is used."""
manager = await auth.auth_manager_from_config(hass, [], [])
user = MockUser().add_to_auth_manager(manager)
now = dt_util.utcnow()
with freeze_time(now):
refresh_token = await manager.async_create_refresh_token(user, CLIENT_ID)
assert (
refresh_token.expire_at
== now.timestamp() + timedelta(days=90).total_seconds()
)
with freeze_time(now + timedelta(days=30)):
async_fire_time_changed(hass, now + timedelta(days=30))
await hass.async_block_till_done()
assert manager.async_create_access_token(refresh_token)
await hass.async_block_till_done()
assert (
refresh_token.expire_at
== now.timestamp()
+ timedelta(days=30).total_seconds()
+ timedelta(days=90).total_seconds()
)
2023-02-20 11:42:56 +01:00
async def test_register_revoke_token_callback(mock_hass) -> None:
"""Test that a registered revoke token callback is called."""
manager = await auth.auth_manager_from_config(mock_hass, [], [])
user = MockUser().add_to_auth_manager(manager)
refresh_token = await manager.async_create_refresh_token(user, CLIENT_ID)
called = False
def cb():
nonlocal called
called = True
manager.async_register_revoke_token_callback(refresh_token.id, cb)
manager.async_remove_refresh_token(refresh_token)
assert called
2023-02-20 11:42:56 +01:00
async def test_unregister_revoke_token_callback(mock_hass) -> None:
"""Test that a revoke token callback can be unregistered."""
manager = await auth.auth_manager_from_config(mock_hass, [], [])
user = MockUser().add_to_auth_manager(manager)
refresh_token = await manager.async_create_refresh_token(user, CLIENT_ID)
called = False
def cb():
nonlocal called
called = True
unregister = manager.async_register_revoke_token_callback(refresh_token.id, cb)
unregister()
manager.async_remove_refresh_token(refresh_token)
assert not called
2023-02-20 11:42:56 +01:00
async def test_create_access_token(mock_hass) -> None:
"""Test normal refresh_token's jwt_key keep same after used."""
manager = await auth.auth_manager_from_config(mock_hass, [], [])
user = MockUser().add_to_auth_manager(manager)
refresh_token = await manager.async_create_refresh_token(user, CLIENT_ID)
assert refresh_token.token_type == auth_models.TOKEN_TYPE_NORMAL
jwt_key = refresh_token.jwt_key
access_token = manager.async_create_access_token(refresh_token)
assert access_token is not None
assert refresh_token.jwt_key == jwt_key
2021-09-08 04:59:02 +01:00
jwt_payload = jwt.decode(access_token, jwt_key, algorithms=["HS256"])
assert jwt_payload["iss"] == refresh_token.id
assert (
jwt_payload["exp"] - jwt_payload["iat"] == timedelta(minutes=30).total_seconds()
2019-07-31 12:25:30 -07:00
)
2023-02-20 11:42:56 +01:00
async def test_create_long_lived_access_token(mock_hass) -> None:
"""Test refresh_token's jwt_key changed for long-lived access token."""
manager = await auth.auth_manager_from_config(mock_hass, [], [])
user = MockUser().add_to_auth_manager(manager)
refresh_token = await manager.async_create_refresh_token(
user,
client_name="GPS Logger",
token_type=auth_models.TOKEN_TYPE_LONG_LIVED_ACCESS_TOKEN,
access_token_expiration=timedelta(days=300),
)
assert refresh_token.token_type == auth_models.TOKEN_TYPE_LONG_LIVED_ACCESS_TOKEN
access_token = manager.async_create_access_token(refresh_token)
2021-09-08 04:59:02 +01:00
jwt_payload = jwt.decode(access_token, refresh_token.jwt_key, algorithms=["HS256"])
assert jwt_payload["iss"] == refresh_token.id
assert (
jwt_payload["exp"] - jwt_payload["iat"] == timedelta(days=300).total_seconds()
)
2023-02-20 11:42:56 +01:00
async def test_one_long_lived_access_token_per_refresh_token(mock_hass) -> None:
"""Test one refresh_token can only have one long-lived access token."""
manager = await auth.auth_manager_from_config(mock_hass, [], [])
user = MockUser().add_to_auth_manager(manager)
refresh_token = await manager.async_create_refresh_token(
user,
client_name="GPS Logger",
token_type=auth_models.TOKEN_TYPE_LONG_LIVED_ACCESS_TOKEN,
access_token_expiration=timedelta(days=3000),
)
assert refresh_token.token_type == auth_models.TOKEN_TYPE_LONG_LIVED_ACCESS_TOKEN
access_token = manager.async_create_access_token(refresh_token)
jwt_key = refresh_token.jwt_key
rt = manager.async_validate_access_token(access_token)
assert rt.id == refresh_token.id
with pytest.raises(ValueError):
await manager.async_create_refresh_token(
user,
client_name="GPS Logger",
token_type=auth_models.TOKEN_TYPE_LONG_LIVED_ACCESS_TOKEN,
access_token_expiration=timedelta(days=3000),
)
manager.async_remove_refresh_token(refresh_token)
assert refresh_token.id not in user.refresh_tokens
rt = manager.async_validate_access_token(access_token)
assert rt is None, "Previous issued access token has been invoked"
refresh_token_2 = await manager.async_create_refresh_token(
user,
client_name="GPS Logger",
token_type=auth_models.TOKEN_TYPE_LONG_LIVED_ACCESS_TOKEN,
access_token_expiration=timedelta(days=3000),
)
assert refresh_token_2.id != refresh_token.id
assert refresh_token_2.token_type == auth_models.TOKEN_TYPE_LONG_LIVED_ACCESS_TOKEN
access_token_2 = manager.async_create_access_token(refresh_token_2)
jwt_key_2 = refresh_token_2.jwt_key
assert access_token != access_token_2
assert jwt_key != jwt_key_2
rt = manager.async_validate_access_token(access_token_2)
2021-09-08 04:59:02 +01:00
jwt_payload = jwt.decode(access_token_2, rt.jwt_key, algorithms=["HS256"])
assert jwt_payload["iss"] == refresh_token_2.id
assert (
jwt_payload["exp"] - jwt_payload["iat"] == timedelta(days=3000).total_seconds()
2019-07-31 12:25:30 -07:00
)
2023-02-20 11:42:56 +01:00
async def test_login_with_auth_module(mock_hass) -> None:
"""Test login as existing user with auth module."""
manager = await auth.auth_manager_from_config(
mock_hass,
[
{
"type": "insecure_example",
"users": [
{
"username": "test-user",
"password": "test-pass",
"name": "Test Name",
}
],
2019-07-31 12:25:30 -07:00
}
],
[
{
"type": "insecure_example",
"data": [{"user_id": "mock-user", "pin": "test-pin"}],
}
2019-07-31 12:25:30 -07:00
],
)
mock_hass.auth = manager
ensure_auth_manager_loaded(manager)
# Add fake user with credentials for example auth provider.
user = MockUser(
id="mock-user", is_owner=False, is_active=False, name="Paulus"
).add_to_auth_manager(manager)
user.credentials.append(
auth_models.Credentials(
id="mock-id",
auth_provider_type="insecure_example",
auth_provider_id=None,
data={"username": "test-user"},
is_new=False,
)
2019-07-31 12:25:30 -07:00
)
step = await manager.login_flow.async_init(("insecure_example", None))
assert step["type"] == data_entry_flow.FlowResultType.FORM
2019-07-31 12:25:30 -07:00
step = await manager.login_flow.async_configure(
step["flow_id"], {"username": "test-user", "password": "test-pass"}
)
# After auth_provider validated, request auth module input form
assert step["type"] == data_entry_flow.FlowResultType.FORM
assert step["step_id"] == "mfa"
step = await manager.login_flow.async_configure(
step["flow_id"], {"pin": "invalid-pin"}
)
# Invalid code error
assert step["type"] == data_entry_flow.FlowResultType.FORM
assert step["step_id"] == "mfa"
assert step["errors"] == {"base": "invalid_code"}
step = await manager.login_flow.async_configure(
step["flow_id"], {"pin": "test-pin"}
)
# Finally passed, get credential
assert step["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
assert step["result"]
assert step["result"].id == "mock-id"
2023-02-20 11:42:56 +01:00
async def test_login_with_multi_auth_module(mock_hass) -> None:
"""Test login as existing user with multiple auth modules."""
manager = await auth.auth_manager_from_config(
mock_hass,
[
{
"type": "insecure_example",
"users": [
{
"username": "test-user",
"password": "test-pass",
"name": "Test Name",
}
],
2019-07-31 12:25:30 -07:00
}
],
[
{
"type": "insecure_example",
"data": [{"user_id": "mock-user", "pin": "test-pin"}],
},
{
"type": "insecure_example",
"id": "module2",
"data": [{"user_id": "mock-user", "pin": "test-pin2"}],
},
2019-07-31 12:25:30 -07:00
],
)
mock_hass.auth = manager
ensure_auth_manager_loaded(manager)
# Add fake user with credentials for example auth provider.
user = MockUser(
id="mock-user", is_owner=False, is_active=False, name="Paulus"
).add_to_auth_manager(manager)
user.credentials.append(
auth_models.Credentials(
id="mock-id",
auth_provider_type="insecure_example",
auth_provider_id=None,
data={"username": "test-user"},
is_new=False,
)
2019-07-31 12:25:30 -07:00
)
step = await manager.login_flow.async_init(("insecure_example", None))
assert step["type"] == data_entry_flow.FlowResultType.FORM
2019-07-31 12:25:30 -07:00
step = await manager.login_flow.async_configure(
step["flow_id"], {"username": "test-user", "password": "test-pass"}
)
# After auth_provider validated, request select auth module
assert step["type"] == data_entry_flow.FlowResultType.FORM
assert step["step_id"] == "select_mfa_module"
step = await manager.login_flow.async_configure(
step["flow_id"], {"multi_factor_auth_module": "module2"}
)
assert step["type"] == data_entry_flow.FlowResultType.FORM
assert step["step_id"] == "mfa"
step = await manager.login_flow.async_configure(
step["flow_id"], {"pin": "test-pin2"}
)
# Finally passed, get credential
assert step["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
assert step["result"]
assert step["result"].id == "mock-id"
2023-02-20 11:42:56 +01:00
async def test_auth_module_expired_session(mock_hass) -> None:
"""Test login as existing user."""
manager = await auth.auth_manager_from_config(
mock_hass,
[
{
"type": "insecure_example",
"users": [
{
"username": "test-user",
"password": "test-pass",
"name": "Test Name",
}
],
2019-07-31 12:25:30 -07:00
}
],
[
{
"type": "insecure_example",
"data": [{"user_id": "mock-user", "pin": "test-pin"}],
}
2019-07-31 12:25:30 -07:00
],
)
mock_hass.auth = manager
ensure_auth_manager_loaded(manager)
# Add fake user with credentials for example auth provider.
user = MockUser(
id="mock-user", is_owner=False, is_active=False, name="Paulus"
).add_to_auth_manager(manager)
user.credentials.append(
auth_models.Credentials(
id="mock-id",
auth_provider_type="insecure_example",
auth_provider_id=None,
data={"username": "test-user"},
is_new=False,
)
2019-07-31 12:25:30 -07:00
)
step = await manager.login_flow.async_init(("insecure_example", None))
assert step["type"] == data_entry_flow.FlowResultType.FORM
2019-07-31 12:25:30 -07:00
step = await manager.login_flow.async_configure(
step["flow_id"], {"username": "test-user", "password": "test-pass"}
)
2019-07-31 12:25:30 -07:00
assert step["type"] == data_entry_flow.FlowResultType.FORM
assert step["step_id"] == "mfa"
2019-07-31 12:25:30 -07:00
with freeze_time(dt_util.utcnow() + MFA_SESSION_EXPIRATION):
step = await manager.login_flow.async_configure(
step["flow_id"], {"pin": "test-pin"}
)
# login flow abort due session timeout
assert step["type"] == data_entry_flow.FlowResultType.ABORT
assert step["reason"] == "login_expired"
2023-02-20 11:42:56 +01:00
async def test_enable_mfa_for_user(
hass: HomeAssistant, hass_storage: dict[str, Any]
) -> None:
"""Test enable mfa module for user."""
manager = await auth.auth_manager_from_config(
hass,
[
{
"type": "insecure_example",
"users": [{"username": "test-user", "password": "test-pass"}],
}
],
[{"type": "insecure_example", "data": []}],
)
2019-07-31 12:25:30 -07:00
step = await manager.login_flow.async_init(("insecure_example", None))
step = await manager.login_flow.async_configure(
step["flow_id"], {"username": "test-user", "password": "test-pass"}
)
credential = step["result"]
user = await manager.async_get_or_create_user(credential)
assert user is not None
# new user don't have mfa enabled
modules = await manager.async_get_enabled_mfa(user)
assert len(modules) == 0
module = manager.get_auth_mfa_module("insecure_example")
# mfa module don't have data
assert bool(module._data) is False
# test enable mfa for user
await manager.async_enable_user_mfa(user, "insecure_example", {"pin": "test-pin"})
assert len(module._data) == 1
assert module._data[0] == {"user_id": user.id, "pin": "test-pin"}
# test get enabled mfa
modules = await manager.async_get_enabled_mfa(user)
assert len(modules) == 1
assert "insecure_example" in modules
# re-enable mfa for user will override
await manager.async_enable_user_mfa(
user, "insecure_example", {"pin": "test-pin-new"}
)
assert len(module._data) == 1
assert module._data[0] == {"user_id": user.id, "pin": "test-pin-new"}
modules = await manager.async_get_enabled_mfa(user)
assert len(modules) == 1
assert "insecure_example" in modules
# system user cannot enable mfa
system_user = await manager.async_create_system_user("system-user")
with pytest.raises(ValueError):
await manager.async_enable_user_mfa(
system_user, "insecure_example", {"pin": "test-pin"}
)
assert len(module._data) == 1
modules = await manager.async_get_enabled_mfa(system_user)
assert len(modules) == 0
# disable mfa for user
await manager.async_disable_user_mfa(user, "insecure_example")
assert bool(module._data) is False
# test get enabled mfa
modules = await manager.async_get_enabled_mfa(user)
assert len(modules) == 0
# disable mfa for user don't enabled just silent fail
await manager.async_disable_user_mfa(user, "insecure_example")
2018-10-11 17:06:51 +02:00
async def test_async_remove_user(hass: HomeAssistant) -> None:
2018-10-11 17:06:51 +02:00
"""Test removing a user."""
events = async_capture_events(hass, "user_removed")
2018-10-11 17:06:51 +02:00
manager = await auth.auth_manager_from_config(
hass,
[
{
"type": "insecure_example",
"users": [
{
"username": "test-user",
"password": "test-pass",
"name": "Test Name",
2019-07-31 12:25:30 -07:00
}
],
2018-10-11 17:06:51 +02:00
}
2019-07-31 12:25:30 -07:00
],
2018-10-11 17:06:51 +02:00
[],
)
hass.auth = manager
ensure_auth_manager_loaded(manager)
# Add fake user with credentials for example auth provider.
user = MockUser(
id="mock-user", is_owner=False, is_active=False, name="Paulus"
).add_to_auth_manager(manager)
user.credentials.append(
auth_models.Credentials(
id="mock-id",
auth_provider_type="insecure_example",
auth_provider_id=None,
data={"username": "test-user"},
is_new=False,
2019-07-31 12:25:30 -07:00
)
2018-10-11 17:06:51 +02:00
)
assert len(user.credentials) == 1
await hass.auth.async_remove_user(user)
assert len(await manager.async_get_users()) == 0
assert len(user.credentials) == 0
await hass.async_block_till_done()
assert len(events) == 1
assert events[0].data["user_id"] == user.id
async def test_async_remove_user_fail_if_remove_credential_fails(
2023-02-20 11:42:56 +01:00
hass: HomeAssistant, hass_admin_user: MockUser, hass_admin_credential: Credentials
) -> None:
"""Test removing a user."""
await hass.auth.async_link_user(hass_admin_user, hass_admin_credential)
with (
patch.object(hass.auth, "async_remove_credentials", side_effect=ValueError),
pytest.raises(ValueError),
):
await hass.auth.async_remove_user(hass_admin_user)
2023-02-20 11:42:56 +01:00
async def test_new_users(mock_hass) -> None:
"""Test newly created users."""
manager = await auth.auth_manager_from_config(
mock_hass,
[
{
"type": "insecure_example",
"users": [
{
"username": "test-user",
"password": "test-pass",
"name": "Test Name",
},
{
"username": "test-user-2",
"password": "test-pass",
"name": "Test Name",
},
{
"username": "test-user-3",
"password": "test-pass",
"name": "Test Name",
},
2019-07-31 12:25:30 -07:00
],
}
2019-07-31 12:25:30 -07:00
],
[],
)
ensure_auth_manager_loaded(manager)
user = await manager.async_create_user("Hello")
# first user in the system is owner and admin
assert user.is_owner
assert user.is_admin
2021-11-29 14:01:03 -08:00
assert not user.local_only
assert user.groups == []
user = await manager.async_create_user("Hello 2")
assert not user.is_admin
assert user.groups == []
2021-11-29 14:01:03 -08:00
user = await manager.async_create_user(
"Hello 3", group_ids=["system-admin"], local_only=True
)
assert user.is_admin
assert user.groups[0].id == "system-admin"
2021-11-29 14:01:03 -08:00
assert user.local_only
user_cred = await manager.async_get_or_create_user(
auth_models.Credentials(
id="mock-id",
auth_provider_type="insecure_example",
auth_provider_id=None,
data={"username": "test-user"},
is_new=True,
2019-07-31 12:25:30 -07:00
)
)
assert user_cred.is_admin
2023-02-20 11:42:56 +01:00
async def test_rename_does_not_change_refresh_token(mock_hass) -> None:
"""Test that we can rename without changing refresh token."""
manager = await auth.auth_manager_from_config(mock_hass, [], [])
user = MockUser().add_to_auth_manager(manager)
await manager.async_create_refresh_token(user, CLIENT_ID)
assert len(list(user.refresh_tokens.values())) == 1
token_before = list(user.refresh_tokens.values())[0]
await manager.async_update_user(user, name="new name")
assert user.name == "new name"
assert len(list(user.refresh_tokens.values())) == 1
token_after = list(user.refresh_tokens.values())[0]
assert token_before == token_after
2022-06-09 17:49:02 -10:00
async def test_event_user_updated_fires(hass: HomeAssistant) -> None:
2022-06-09 17:49:02 -10:00
"""Test the user updated event fires."""
manager = await auth.auth_manager_from_config(hass, [], [])
user = MockUser().add_to_auth_manager(manager)
await manager.async_create_refresh_token(user, CLIENT_ID)
assert len(list(user.refresh_tokens.values())) == 1
events = async_capture_events(hass, EVENT_USER_UPDATED)
await manager.async_update_user(user, name="new name")
assert user.name == "new name"
await hass.async_block_till_done()
assert len(events) == 1
2023-03-22 16:03:41 -10:00
async def test_access_token_with_invalid_signature(mock_hass) -> None:
"""Test rejecting access tokens with an invalid signature."""
manager = await auth.auth_manager_from_config(mock_hass, [], [])
user = MockUser().add_to_auth_manager(manager)
refresh_token = await manager.async_create_refresh_token(
user,
client_name="Good Client",
token_type=auth_models.TOKEN_TYPE_LONG_LIVED_ACCESS_TOKEN,
access_token_expiration=timedelta(days=3000),
)
assert refresh_token.token_type == auth_models.TOKEN_TYPE_LONG_LIVED_ACCESS_TOKEN
access_token = manager.async_create_access_token(refresh_token)
rt = manager.async_validate_access_token(access_token)
2023-03-22 16:03:41 -10:00
assert rt.id == refresh_token.id
# Now we corrupt the signature
header, payload, signature = access_token.split(".")
invalid_signature = "a" * len(signature)
invalid_token = f"{header}.{payload}.{invalid_signature}"
assert access_token != invalid_token
result = manager.async_validate_access_token(invalid_token)
2023-03-22 16:03:41 -10:00
assert result is None
async def test_access_token_with_null_signature(mock_hass) -> None:
"""Test rejecting access tokens with a null signature."""
manager = await auth.auth_manager_from_config(mock_hass, [], [])
user = MockUser().add_to_auth_manager(manager)
refresh_token = await manager.async_create_refresh_token(
user,
client_name="Good Client",
token_type=auth_models.TOKEN_TYPE_LONG_LIVED_ACCESS_TOKEN,
access_token_expiration=timedelta(days=3000),
)
assert refresh_token.token_type == auth_models.TOKEN_TYPE_LONG_LIVED_ACCESS_TOKEN
access_token = manager.async_create_access_token(refresh_token)
rt = manager.async_validate_access_token(access_token)
2023-03-22 16:03:41 -10:00
assert rt.id == refresh_token.id
# Now we make the signature all nulls
header, payload, signature = access_token.split(".")
invalid_signature = "\0" * len(signature)
invalid_token = f"{header}.{payload}.{invalid_signature}"
assert access_token != invalid_token
result = manager.async_validate_access_token(invalid_token)
2023-03-22 16:03:41 -10:00
assert result is None
async def test_access_token_with_empty_signature(mock_hass) -> None:
"""Test rejecting access tokens with an empty signature."""
manager = await auth.auth_manager_from_config(mock_hass, [], [])
user = MockUser().add_to_auth_manager(manager)
refresh_token = await manager.async_create_refresh_token(
user,
client_name="Good Client",
token_type=auth_models.TOKEN_TYPE_LONG_LIVED_ACCESS_TOKEN,
access_token_expiration=timedelta(days=3000),
)
assert refresh_token.token_type == auth_models.TOKEN_TYPE_LONG_LIVED_ACCESS_TOKEN
access_token = manager.async_create_access_token(refresh_token)
rt = manager.async_validate_access_token(access_token)
2023-03-22 16:03:41 -10:00
assert rt.id == refresh_token.id
# Now we make the signature all nulls
header, payload, _ = access_token.split(".")
invalid_token = f"{header}.{payload}."
assert access_token != invalid_token
result = manager.async_validate_access_token(invalid_token)
2023-03-22 16:03:41 -10:00
assert result is None
async def test_access_token_with_empty_key(mock_hass) -> None:
"""Test rejecting access tokens with an empty key."""
manager = await auth.auth_manager_from_config(mock_hass, [], [])
user = MockUser().add_to_auth_manager(manager)
refresh_token = await manager.async_create_refresh_token(
user,
client_name="Good Client",
token_type=auth_models.TOKEN_TYPE_LONG_LIVED_ACCESS_TOKEN,
access_token_expiration=timedelta(days=3000),
)
assert refresh_token.token_type == auth_models.TOKEN_TYPE_LONG_LIVED_ACCESS_TOKEN
access_token = manager.async_create_access_token(refresh_token)
manager.async_remove_refresh_token(refresh_token)
2023-03-22 16:03:41 -10:00
# Now remove the token from the keyring
# so we will get an empty key
assert manager.async_validate_access_token(access_token) is None
2023-03-22 16:03:41 -10:00
async def test_reject_access_token_with_impossible_large_size(mock_hass) -> None:
"""Test rejecting access tokens with impossible sizes."""
manager = await auth.auth_manager_from_config(mock_hass, [], [])
assert manager.async_validate_access_token("a" * 10000) is None
2023-03-22 16:03:41 -10:00
async def test_reject_token_with_invalid_json_payload(mock_hass) -> None:
"""Test rejecting access tokens with invalid json payload."""
jws = jwt.PyJWS()
token_with_invalid_json = jws.encode(
b"invalid", b"invalid", "HS256", {"alg": "HS256", "typ": "JWT"}
)
manager = await auth.auth_manager_from_config(mock_hass, [], [])
assert manager.async_validate_access_token(token_with_invalid_json) is None
2023-03-22 16:03:41 -10:00
async def test_reject_token_with_not_dict_json_payload(mock_hass) -> None:
"""Test rejecting access tokens with not a dict json payload."""
jws = jwt.PyJWS()
token_not_a_dict_json = jws.encode(
b'["invalid"]', b"invalid", "HS256", {"alg": "HS256", "typ": "JWT"}
)
manager = await auth.auth_manager_from_config(mock_hass, [], [])
assert manager.async_validate_access_token(token_not_a_dict_json) is None
2023-03-22 16:03:41 -10:00
async def test_access_token_that_expires_soon(mock_hass) -> None:
"""Test access token from refresh token that expires very soon."""
now = dt_util.utcnow()
manager = await auth.auth_manager_from_config(mock_hass, [], [])
user = MockUser().add_to_auth_manager(manager)
refresh_token = await manager.async_create_refresh_token(
user,
client_name="Token that expires very soon",
token_type=auth_models.TOKEN_TYPE_LONG_LIVED_ACCESS_TOKEN,
access_token_expiration=timedelta(seconds=1),
)
assert refresh_token.token_type == auth_models.TOKEN_TYPE_LONG_LIVED_ACCESS_TOKEN
access_token = manager.async_create_access_token(refresh_token)
rt = manager.async_validate_access_token(access_token)
2023-03-22 16:03:41 -10:00
assert rt.id == refresh_token.id
with freeze_time(now + timedelta(minutes=1)):
assert manager.async_validate_access_token(access_token) is None
2023-03-22 16:03:41 -10:00
async def test_access_token_from_the_future(mock_hass) -> None:
"""Test we reject an access token from the future."""
now = dt_util.utcnow()
manager = await auth.auth_manager_from_config(mock_hass, [], [])
user = MockUser().add_to_auth_manager(manager)
with freeze_time(now + timedelta(days=365)):
refresh_token = await manager.async_create_refresh_token(
user,
client_name="Token that expires very soon",
token_type=auth_models.TOKEN_TYPE_LONG_LIVED_ACCESS_TOKEN,
access_token_expiration=timedelta(days=10),
)
assert (
refresh_token.token_type == auth_models.TOKEN_TYPE_LONG_LIVED_ACCESS_TOKEN
)
access_token = manager.async_create_access_token(refresh_token)
assert manager.async_validate_access_token(access_token) is None
2023-03-22 16:03:41 -10:00
with freeze_time(now + timedelta(days=365)):
rt = manager.async_validate_access_token(access_token)
2023-03-22 16:03:41 -10:00
assert rt.id == refresh_token.id