Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-12-24 21:06:19 +00:00
Code Activity

Remove direct authentication via trusted networks or API password (#27656)

Browse Source 
* Remove direct authentication via trusted networks and API password

* Fix tests
This commit is contained in:
Paulus Schoutsen
2019-10-14 14:56:45 -07:00
committed by GitHub
parent 97478d1ef4
commit 3231e22ddf
33 changed files with 114 additions and 423 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
tests/components/websocket_api/__init__.py
View File

@@ -1,2 +1 @@
"""Tests for the websocket API."""
API_PASSWORD = "test-password"

8
tests/components/websocket_api/conftest.py
View File

@@ -5,8 +5,6 @@ from homeassistant.setup import async_setup_component
from homeassistant.components.websocket_api.http import URL
from homeassistant.components.websocket_api.auth import TYPE_AUTH_REQUIRED
from . import API_PASSWORD
@pytest.fixture
def websocket_client(hass, hass_ws_client, hass_access_token):
@@ -17,11 +15,7 @@ def websocket_client(hass, hass_ws_client, hass_access_token):
@pytest.fixture
def no_auth_websocket_client(hass, loop, aiohttp_client):
"""Websocket connection that requires authentication."""
assert loop.run_until_complete(
async_setup_component(
hass, "websocket_api", {"http": {"api_password": API_PASSWORD}}
)
)
assert loop.run_until_complete(async_setup_component(hass, "websocket_api", {}))
client = loop.run_until_complete(aiohttp_client(hass.http.app))
ws = loop.run_until_complete(client.ws_connect(URL))

72
tests/components/websocket_api/test_auth.py
View File

@@ -17,21 +17,10 @@ from homeassistant.setup import async_setup_component
from tests.common import mock_coro
from . import API_PASSWORD
async def test_auth_via_msg(no_auth_websocket_client, legacy_auth):
"""Test authenticating."""
await no_auth_websocket_client.send_json(
{"type": TYPE_AUTH, "api_password": API_PASSWORD}
)
msg = await no_auth_websocket_client.receive_json()
assert msg["type"] == TYPE_AUTH_OK
async def test_auth_events(hass, no_auth_websocket_client, legacy_auth):
async def test_auth_events(
hass, no_auth_websocket_client, legacy_auth, hass_access_token
):
"""Test authenticating."""
connected_evt = []
hass.helpers.dispatcher.async_dispatcher_connect(
@@ -42,7 +31,7 @@ async def test_auth_events(hass, no_auth_websocket_client, legacy_auth):
SIGNAL_WEBSOCKET_DISCONNECTED, lambda: disconnected_evt.append(1)
)
await test_auth_via_msg(no_auth_websocket_client, legacy_auth)
await test_auth_active_with_token(hass, no_auth_websocket_client, hass_access_token)
assert len(connected_evt) == 1
assert not disconnected_evt
@@ -60,7 +49,7 @@ async def test_auth_via_msg_incorrect_pass(no_auth_websocket_client):
return_value=mock_coro(),
) as mock_process_wrong_login:
await no_auth_websocket_client.send_json(
{"type": TYPE_AUTH, "api_password": API_PASSWORD + "wrong"}
{"type": TYPE_AUTH, "api_password": "wrong"}
)
msg = await no_auth_websocket_client.receive_json()
@@ -110,31 +99,25 @@ async def test_pre_auth_only_auth_allowed(no_auth_websocket_client):
assert msg["message"].startswith("Auth message incorrectly formatted")
async def test_auth_active_with_token(hass, aiohttp_client, hass_access_token):
async def test_auth_active_with_token(
hass, no_auth_websocket_client, hass_access_token
):
"""Test authenticating with a token."""
assert await async_setup_component(
hass, "websocket_api", {"http": {"api_password": API_PASSWORD}}
assert await async_setup_component(hass, "websocket_api", {})
await no_auth_websocket_client.send_json(
{"type": TYPE_AUTH, "access_token": hass_access_token}
)
client = await aiohttp_client(hass.http.app)
async with client.ws_connect(URL) as ws:
auth_msg = await ws.receive_json()
assert auth_msg["type"] == TYPE_AUTH_REQUIRED
await ws.send_json({"type": TYPE_AUTH, "access_token": hass_access_token})
auth_msg = await ws.receive_json()
assert auth_msg["type"] == TYPE_AUTH_OK
auth_msg = await no_auth_websocket_client.receive_json()
assert auth_msg["type"] == TYPE_AUTH_OK
async def test_auth_active_user_inactive(hass, aiohttp_client, hass_access_token):
"""Test authenticating with a token."""
refresh_token = await hass.auth.async_validate_access_token(hass_access_token)
refresh_token.user.is_active = False
assert await async_setup_component(
hass, "websocket_api", {"http": {"api_password": API_PASSWORD}}
)
assert await async_setup_component(hass, "websocket_api", {})
client = await aiohttp_client(hass.http.app)
@@ -150,9 +133,7 @@ async def test_auth_active_user_inactive(hass, aiohttp_client, hass_access_token
async def test_auth_active_with_password_not_allow(hass, aiohttp_client):
"""Test authenticating with a token."""
assert await async_setup_component(
hass, "websocket_api", {"http": {"api_password": API_PASSWORD}}
)
assert await async_setup_component(hass, "websocket_api", {})
client = await aiohttp_client(hass.http.app)
@@ -160,7 +141,7 @@ async def test_auth_active_with_password_not_allow(hass, aiohttp_client):
auth_msg = await ws.receive_json()
assert auth_msg["type"] == TYPE_AUTH_REQUIRED
await ws.send_json({"type": TYPE_AUTH, "api_password": API_PASSWORD})
await ws.send_json({"type": TYPE_AUTH, "api_password": "some-password"})
auth_msg = await ws.receive_json()
assert auth_msg["type"] == TYPE_AUTH_INVALID
@@ -168,28 +149,23 @@ async def test_auth_active_with_password_not_allow(hass, aiohttp_client):
async def test_auth_legacy_support_with_password(hass, aiohttp_client, legacy_auth):
"""Test authenticating with a token."""
assert await async_setup_component(
hass, "websocket_api", {"http": {"api_password": API_PASSWORD}}
)
assert await async_setup_component(hass, "websocket_api", {})
client = await aiohttp_client(hass.http.app)
async with client.ws_connect(URL) as ws:
with patch("homeassistant.auth.AuthManager.support_legacy", return_value=True):
auth_msg = await ws.receive_json()
assert auth_msg["type"] == TYPE_AUTH_REQUIRED
auth_msg = await ws.receive_json()
assert auth_msg["type"] == TYPE_AUTH_REQUIRED
await ws.send_json({"type": TYPE_AUTH, "api_password": API_PASSWORD})
await ws.send_json({"type": TYPE_AUTH, "api_password": "some-password"})
auth_msg = await ws.receive_json()
assert auth_msg["type"] == TYPE_AUTH_OK
auth_msg = await ws.receive_json()
assert auth_msg["type"] == TYPE_AUTH_INVALID
async def test_auth_with_invalid_token(hass, aiohttp_client):
"""Test authenticating with a token."""
assert await async_setup_component(
hass, "websocket_api", {"http": {"api_password": API_PASSWORD}}
)
assert await async_setup_component(hass, "websocket_api", {})
client = await aiohttp_client(hass.http.app)

6
tests/components/websocket_api/test_commands.py
View File

@@ -14,8 +14,6 @@ from homeassistant.setup import async_setup_component
from tests.common import async_mock_service
from . import API_PASSWORD
async def test_call_service(hass, websocket_client):
"""Test call service command."""
@@ -250,9 +248,7 @@ async def test_ping(websocket_client):
async def test_call_service_context_with_user(hass, aiohttp_client, hass_access_token):
"""Test that the user is set in the service call context."""
assert await async_setup_component(
hass, "websocket_api", {"http": {"api_password": API_PASSWORD}}
)
assert await async_setup_component(hass, "websocket_api", {})
calls = async_mock_service(hass, "domain_test", "test_service")
client = await aiohttp_client(hass.http.app)

8
tests/components/websocket_api/test_sensor.py
View File

@@ -3,10 +3,12 @@
from homeassistant.bootstrap import async_setup_component
from tests.common import assert_setup_component
from .test_auth import test_auth_via_msg
from .test_auth import test_auth_active_with_token
async def test_websocket_api(hass, no_auth_websocket_client, legacy_auth):
async def test_websocket_api(
hass, no_auth_websocket_client, hass_access_token, legacy_auth
):
"""Test API streams."""
with assert_setup_component(1):
await async_setup_component(
@@ -16,7 +18,7 @@ async def test_websocket_api(hass, no_auth_websocket_client, legacy_auth):
state = hass.states.get("sensor.connected_clients")
assert state.state == "0"
await test_auth_via_msg(no_auth_websocket_client, legacy_auth)
await test_auth_active_with_token(hass, no_auth_websocket_client, hass_access_token)
state = hass.states.get("sensor.connected_clients")
assert state.state == "1"