Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Always set hass_user (#18844)

pull/18855/head
Paulus Schoutsen 2018-11-30 17:32:47 +01:00 committed by GitHub
parent 8f50180598
commit d014517ce2
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 79 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
homeassistant/components/http/auth.py
View File

@ -45,6 +45,7 @@ def setup_auth(app, trusted_networks, use_auth,
support_legacy=False, api_password=None):
"""Create auth middleware for the app."""
old_auth_warning = set()
legacy_auth = (not use_auth or support_legacy) and api_password
@middleware
async def auth_middleware(request, handler):
@ -60,7 +61,6 @@ def setup_auth(app, trusted_networks, use_auth,
request.path, request[KEY_REAL_IP])
old_auth_warning.add(request.path)
legacy_auth = (not use_auth or support_legacy) and api_password
if (hdrs.AUTHORIZATION in request.headers and
await async_validate_auth_header(
request, api_password if legacy_auth else None)):
@ -91,6 +91,11 @@ def setup_auth(app, trusted_networks, use_auth,
app['hass'])
elif _is_trusted_ip(request, trusted_networks):
users = await app['hass'].auth.async_get_users()
for user in users:
if user.is_owner:
request['hass_user'] = user
break
authenticated = True
elif not use_auth and api_password is None:
@ -136,8 +141,9 @@ async def async_validate_auth_header(request, api_password=None):
# If no space in authorization header
return False
hass = request.app['hass']
if auth_type == 'Bearer':
hass = request.app['hass']
refresh_token = await hass.auth.async_validate_access_token(auth_val)
if refresh_token is None:
return False
@ -157,8 +163,12 @@ async def async_validate_auth_header(request, api_password=None):
if username != 'homeassistant':
return False
return hmac.compare_digest(api_password.encode('utf-8'),
password.encode('utf-8'))
if not hmac.compare_digest(api_password.encode('utf-8'),
password.encode('utf-8')):
return False
request['hass_user'] = await legacy_api_password.async_get_user(hass)
return True
return False

6
tests/components/conftest.py
View File

@ -88,6 +88,12 @@ def hass_access_token(hass, hass_admin_user):
yield hass.auth.async_create_access_token(refresh_token)
@pytest.fixture
def hass_owner_user(hass, local_auth):
"""Return a Home Assistant admin user."""
return MockUser(is_owner=True).add_to_hass(hass)
@pytest.fixture
def hass_admin_user(hass, local_auth):
"""Return a Home Assistant admin user."""

65
tests/components/http/test_auth.py
View File

@ -7,6 +7,7 @@ import pytest
from aiohttp import BasicAuth, web
from aiohttp.web_exceptions import HTTPUnauthorized
from homeassistant.auth.providers import legacy_api_password
from homeassistant.components.http.auth import setup_auth, async_sign_path
from homeassistant.components.http.const import KEY_AUTHENTICATED
from homeassistant.components.http.real_ip import setup_real_ip
@ -84,29 +85,40 @@ async def test_access_without_password(app, aiohttp_client):
async def test_access_with_password_in_header(app, aiohttp_client,
legacy_auth):
legacy_auth, hass):
"""Test access with password in header."""
setup_auth(app, [], False, api_password=API_PASSWORD)
client = await aiohttp_client(app)
user = await legacy_api_password.async_get_user(hass)
req = await client.get(
'/', headers={HTTP_HEADER_HA_AUTH: API_PASSWORD})
assert req.status == 200
assert await req.json() == {
'refresh_token_id': None,
'user_id': user.id,
}
req = await client.get(
'/', headers={HTTP_HEADER_HA_AUTH: 'wrong-pass'})
assert req.status == 401
async def test_access_with_password_in_query(app, aiohttp_client, legacy_auth):
async def test_access_with_password_in_query(app, aiohttp_client, legacy_auth,
hass):
"""Test access with password in URL."""
setup_auth(app, [], False, api_password=API_PASSWORD)
client = await aiohttp_client(app)
user = await legacy_api_password.async_get_user(hass)
resp = await client.get('/', params={
'api_password': API_PASSWORD
})
assert resp.status == 200
assert await resp.json() == {
'refresh_token_id': None,
'user_id': user.id,
}
resp = await client.get('/')
assert resp.status == 401
@ -117,15 +129,20 @@ async def test_access_with_password_in_query(app, aiohttp_client, legacy_auth):
assert resp.status == 401
async def test_basic_auth_works(app, aiohttp_client):
async def test_basic_auth_works(app, aiohttp_client, hass, legacy_auth):
"""Test access with basic authentication."""
setup_auth(app, [], False, api_password=API_PASSWORD)
client = await aiohttp_client(app)
user = await legacy_api_password.async_get_user(hass)
req = await client.get(
'/',
auth=BasicAuth('homeassistant', API_PASSWORD))
assert req.status == 200
assert await req.json() == {
'refresh_token_id': None,
'user_id': user.id,
}
req = await client.get(
'/',
@ -145,7 +162,7 @@ async def test_basic_auth_works(app, aiohttp_client):
assert req.status == 401
async def test_access_with_trusted_ip(app2, aiohttp_client):
async def test_access_with_trusted_ip(app2, aiohttp_client, hass_owner_user):
"""Test access with an untrusted ip address."""
setup_auth(app2, TRUSTED_NETWORKS, False, api_password='some-pass')
@ -163,6 +180,10 @@ async def test_access_with_trusted_ip(app2, aiohttp_client):
resp = await client.get('/')
assert resp.status == 200, \
"{} should be trusted".format(remote_addr)
assert await resp.json() == {
'refresh_token_id': None,
'user_id': hass_owner_user.id,
}
async def test_auth_active_access_with_access_token_in_header(
@ -171,18 +192,32 @@ async def test_auth_active_access_with_access_token_in_header(
token = hass_access_token
setup_auth(app, [], True, api_password=None)
client = await aiohttp_client(app)
refresh_token = await hass.auth.async_validate_access_token(
hass_access_token)
req = await client.get(
'/', headers={'Authorization': 'Bearer {}'.format(token)})
assert req.status == 200
assert await req.json() == {
'refresh_token_id': refresh_token.id,
'user_id': refresh_token.user.id,
}
req = await client.get(
'/', headers={'AUTHORIZATION': 'Bearer {}'.format(token)})
assert req.status == 200
assert await req.json() == {
'refresh_token_id': refresh_token.id,
'user_id': refresh_token.user.id,
}
req = await client.get(
'/', headers={'authorization': 'Bearer {}'.format(token)})
assert req.status == 200
assert await req.json() == {
'refresh_token_id': refresh_token.id,
'user_id': refresh_token.user.id,
}
req = await client.get(
'/', headers={'Authorization': token})
@ -200,7 +235,8 @@ async def test_auth_active_access_with_access_token_in_header(
assert req.status == 401
async def test_auth_active_access_with_trusted_ip(app2, aiohttp_client):
async def test_auth_active_access_with_trusted_ip(app2, aiohttp_client,
hass_owner_user):
"""Test access with an untrusted ip address."""
setup_auth(app2, TRUSTED_NETWORKS, True, api_password=None)
@ -218,6 +254,10 @@ async def test_auth_active_access_with_trusted_ip(app2, aiohttp_client):
resp = await client.get('/')
assert resp.status == 200, \
"{} should be trusted".format(remote_addr)
assert await resp.json() == {
'refresh_token_id': None,
'user_id': hass_owner_user.id,
}
async def test_auth_active_blocked_api_password_access(
@ -242,24 +282,37 @@ async def test_auth_active_blocked_api_password_access(
async def test_auth_legacy_support_api_password_access(
app, aiohttp_client, legacy_auth):
app, aiohttp_client, legacy_auth, hass):
"""Test access using api_password if auth.support_legacy."""
setup_auth(app, [], True, support_legacy=True, api_password=API_PASSWORD)
client = await aiohttp_client(app)
user = await legacy_api_password.async_get_user(hass)
req = await client.get(
'/', headers={HTTP_HEADER_HA_AUTH: API_PASSWORD})
assert req.status == 200
assert await req.json() == {
'refresh_token_id': None,
'user_id': user.id,
}
resp = await client.get('/', params={
'api_password': API_PASSWORD
})
assert resp.status == 200
assert await resp.json() == {
'refresh_token_id': None,
'user_id': user.id,
}
req = await client.get(
'/',
auth=BasicAuth('homeassistant', API_PASSWORD))
assert req.status == 200
assert await req.json() == {
'refresh_token_id': None,
'user_id': user.id,
}
async def test_auth_access_signed_path(