Reorganize HTTP component (#4575)

* Move HTTP to own folder

* Break HTTP into middlewares

* Lint

* Split tests per middleware

* Clean up HTTP tests

* Make HomeAssistantViews more stateless

* Lint

* Make HTTP setup async
This commit is contained in:
Paulus Schoutsen
2016-11-25 13:04:06 -08:00
committed by GitHub
parent 58b85b2e0e
commit 32ffd006fa
35 changed files with 1318 additions and 1084 deletions
+16 -1
View File
@@ -10,6 +10,8 @@ import logging
import threading
from contextlib import contextmanager
from aiohttp import web
from homeassistant import core as ha, loader
from homeassistant.bootstrap import (
setup_component, async_prepare_setup_component)
@@ -22,6 +24,9 @@ from homeassistant.const import (
EVENT_STATE_CHANGED, EVENT_PLATFORM_DISCOVERED, ATTR_SERVICE,
ATTR_DISCOVERED, SERVER_PORT)
from homeassistant.components import sun, mqtt
from homeassistant.components.http.auth import auth_middleware
from homeassistant.components.http.const import (
KEY_USE_X_FORWARDED_FOR, KEY_BANS_ENABLED)
_TEST_INSTANCE_PORT = SERVER_PORT
_LOGGER = logging.getLogger(__name__)
@@ -210,13 +215,23 @@ def mock_http_component(hass):
"""Store registered view."""
if isinstance(view, type):
# Instantiate the view, if needed
view = view(hass)
view = view()
hass.http.views[view.name] = view
hass.http.register_view = mock_register_view
def mock_http_component_app(hass):
"""Create an aiohttp.web.Application instance for testing."""
hass.http.api_password = None
app = web.Application(middlewares=[auth_middleware], loop=hass.loop)
app['hass'] = hass
app[KEY_USE_X_FORWARDED_FOR] = False
app[KEY_BANS_ENABLED] = False
return app
def mock_mqtt_component(hass):
"""Mock the MQTT component."""
with mock.patch('homeassistant.components.mqtt.MQTT') as mock_mqtt:
+1 -1
View File
@@ -27,8 +27,8 @@ def test_fetching_url(aioclient_mock, hass, test_client):
resp = yield from client.get('/api/camera_proxy/camera.config_test')
assert aioclient_mock.call_count == 1
assert resp.status == 200
assert aioclient_mock.call_count == 1
body = yield from resp.text()
assert body == 'hello world'
+1
View File
@@ -0,0 +1 @@
"""Tests for the HTTP component."""
+169
View File
@@ -0,0 +1,169 @@
"""The tests for the Home Assistant HTTP component."""
# pylint: disable=protected-access
import logging
from ipaddress import ip_address, ip_network
from unittest.mock import patch
import requests
from homeassistant import bootstrap, const
import homeassistant.components.http as http
from homeassistant.components.http.const import (
KEY_TRUSTED_NETWORKS, KEY_USE_X_FORWARDED_FOR, HTTP_HEADER_X_FORWARDED_FOR)
from tests.common import get_test_instance_port, get_test_home_assistant
API_PASSWORD = 'test1234'
SERVER_PORT = get_test_instance_port()
HTTP_BASE = '127.0.0.1:{}'.format(SERVER_PORT)
HTTP_BASE_URL = 'http://{}'.format(HTTP_BASE)
HA_HEADERS = {
const.HTTP_HEADER_HA_AUTH: API_PASSWORD,
const.HTTP_HEADER_CONTENT_TYPE: const.CONTENT_TYPE_JSON,
}
# Don't add 127.0.0.1/::1 as trusted, as it may interfere with other test cases
TRUSTED_NETWORKS = ['192.0.2.0/24', '2001:DB8:ABCD::/48', '100.64.0.1',
'FD01:DB8::1']
TRUSTED_ADDRESSES = ['100.64.0.1', '192.0.2.100', 'FD01:DB8::1',
'2001:DB8:ABCD::1']
UNTRUSTED_ADDRESSES = ['198.51.100.1', '2001:DB8:FA1::1', '127.0.0.1', '::1']
hass = None
def _url(path=''):
"""Helper method to generate URLs."""
return HTTP_BASE_URL + path
# pylint: disable=invalid-name
def setUpModule():
"""Initialize a Home Assistant server."""
global hass
hass = get_test_home_assistant()
bootstrap.setup_component(
hass, http.DOMAIN, {
http.DOMAIN: {
http.CONF_API_PASSWORD: API_PASSWORD,
http.CONF_SERVER_PORT: SERVER_PORT,
}
}
)
bootstrap.setup_component(hass, 'api')
hass.http.app[KEY_TRUSTED_NETWORKS] = [
ip_network(trusted_network)
for trusted_network in TRUSTED_NETWORKS]
hass.start()
# pylint: disable=invalid-name
def tearDownModule():
"""Stop the Home Assistant server."""
hass.stop()
class TestHttp:
"""Test HTTP component."""
def test_access_denied_without_password(self):
"""Test access without password."""
req = requests.get(_url(const.URL_API))
assert req.status_code == 401
def test_access_denied_with_wrong_password_in_header(self):
"""Test access with wrong password."""
req = requests.get(
_url(const.URL_API),
headers={const.HTTP_HEADER_HA_AUTH: 'wrongpassword'})
assert req.status_code == 401
def test_access_denied_with_x_forwarded_for(self, caplog):
"""Test access denied through the X-Forwarded-For http header."""
hass.http.use_x_forwarded_for = True
for remote_addr in UNTRUSTED_ADDRESSES:
req = requests.get(_url(const.URL_API), headers={
HTTP_HEADER_X_FORWARDED_FOR: remote_addr})
assert req.status_code == 401, \
"{} shouldn't be trusted".format(remote_addr)
def test_access_denied_with_untrusted_ip(self, caplog):
"""Test access with an untrusted ip address."""
for remote_addr in UNTRUSTED_ADDRESSES:
with patch('homeassistant.components.http.'
'util.get_real_ip',
return_value=ip_address(remote_addr)):
req = requests.get(
_url(const.URL_API), params={'api_password': ''})
assert req.status_code == 401, \
"{} shouldn't be trusted".format(remote_addr)
def test_access_with_password_in_header(self, caplog):
"""Test access with password in URL."""
# Hide logging from requests package that we use to test logging
caplog.set_level(
logging.WARNING, logger='requests.packages.urllib3.connectionpool')
req = requests.get(
_url(const.URL_API),
headers={const.HTTP_HEADER_HA_AUTH: API_PASSWORD})
assert req.status_code == 200
logs = caplog.text
assert const.URL_API in logs
assert API_PASSWORD not in logs
def test_access_denied_with_wrong_password_in_url(self):
"""Test access with wrong password."""
req = requests.get(
_url(const.URL_API), params={'api_password': 'wrongpassword'})
assert req.status_code == 401
def test_access_with_password_in_url(self, caplog):
"""Test access with password in URL."""
# Hide logging from requests package that we use to test logging
caplog.set_level(
logging.WARNING, logger='requests.packages.urllib3.connectionpool')
req = requests.get(
_url(const.URL_API), params={'api_password': API_PASSWORD})
assert req.status_code == 200
logs = caplog.text
assert const.URL_API in logs
assert API_PASSWORD not in logs
def test_access_granted_with_x_forwarded_for(self, caplog):
"""Test access denied through the X-Forwarded-For http header."""
hass.http.app[KEY_USE_X_FORWARDED_FOR] = True
for remote_addr in TRUSTED_ADDRESSES:
req = requests.get(_url(const.URL_API), headers={
HTTP_HEADER_X_FORWARDED_FOR: remote_addr})
assert req.status_code == 200, \
"{} should be trusted".format(remote_addr)
def test_access_granted_with_trusted_ip(self, caplog):
"""Test access with trusted addresses."""
for remote_addr in TRUSTED_ADDRESSES:
with patch('homeassistant.components.http.'
'auth.get_real_ip',
return_value=ip_address(remote_addr)):
req = requests.get(
_url(const.URL_API), params={'api_password': ''})
assert req.status_code == 200, \
'{} should be trusted'.format(remote_addr)
+118
View File
@@ -0,0 +1,118 @@
"""The tests for the Home Assistant HTTP component."""
# pylint: disable=protected-access
from ipaddress import ip_address
from unittest.mock import patch, mock_open
import requests
from homeassistant import bootstrap, const
import homeassistant.components.http as http
from homeassistant.components.http.const import (
KEY_BANS_ENABLED, KEY_LOGIN_THRESHOLD, KEY_BANNED_IPS)
from homeassistant.components.http.ban import IpBan, IP_BANS_FILE
from tests.common import get_test_instance_port, get_test_home_assistant
API_PASSWORD = 'test1234'
SERVER_PORT = get_test_instance_port()
HTTP_BASE = '127.0.0.1:{}'.format(SERVER_PORT)
HTTP_BASE_URL = 'http://{}'.format(HTTP_BASE)
HA_HEADERS = {
const.HTTP_HEADER_HA_AUTH: API_PASSWORD,
const.HTTP_HEADER_CONTENT_TYPE: const.CONTENT_TYPE_JSON,
}
BANNED_IPS = ['200.201.202.203', '100.64.0.2']
hass = None
def _url(path=''):
"""Helper method to generate URLs."""
return HTTP_BASE_URL + path
# pylint: disable=invalid-name
def setUpModule():
"""Initialize a Home Assistant server."""
global hass
hass = get_test_home_assistant()
bootstrap.setup_component(
hass, http.DOMAIN, {
http.DOMAIN: {
http.CONF_API_PASSWORD: API_PASSWORD,
http.CONF_SERVER_PORT: SERVER_PORT,
}
}
)
bootstrap.setup_component(hass, 'api')
hass.http.app[KEY_BANNED_IPS] = [IpBan(banned_ip) for banned_ip
in BANNED_IPS]
hass.start()
# pylint: disable=invalid-name
def tearDownModule():
"""Stop the Home Assistant server."""
hass.stop()
class TestHttp:
"""Test HTTP component."""
def test_access_from_banned_ip(self):
"""Test accessing to server from banned IP. Both trusted and not."""
hass.http.app[KEY_BANS_ENABLED] = True
for remote_addr in BANNED_IPS:
with patch('homeassistant.components.http.'
'ban.get_real_ip',
return_value=ip_address(remote_addr)):
req = requests.get(
_url(const.URL_API))
assert req.status_code == 403
def test_access_from_banned_ip_when_ban_is_off(self):
"""Test accessing to server from banned IP when feature is off"""
hass.http.app[KEY_BANS_ENABLED] = False
for remote_addr in BANNED_IPS:
with patch('homeassistant.components.http.'
'ban.get_real_ip',
return_value=ip_address(remote_addr)):
req = requests.get(
_url(const.URL_API),
headers={const.HTTP_HEADER_HA_AUTH: API_PASSWORD})
assert req.status_code == 200
def test_ip_bans_file_creation(self):
"""Testing if banned IP file created"""
hass.http.app[KEY_BANS_ENABLED] = True
hass.http.app[KEY_LOGIN_THRESHOLD] = 1
m = mock_open()
def call_server():
with patch('homeassistant.components.http.'
'ban.get_real_ip',
return_value=ip_address("200.201.202.204")):
print("GETTING API")
return requests.get(
_url(const.URL_API),
headers={const.HTTP_HEADER_HA_AUTH: 'Wrong password'})
with patch('homeassistant.components.http.ban.open', m, create=True):
req = call_server()
assert req.status_code == 401
assert len(hass.http.app[KEY_BANNED_IPS]) == len(BANNED_IPS)
assert m.call_count == 0
req = call_server()
assert req.status_code == 401
assert len(hass.http.app[KEY_BANNED_IPS]) == len(BANNED_IPS) + 1
m.assert_called_once_with(hass.config.path(IP_BANS_FILE), 'a')
req = call_server()
assert req.status_code == 403
assert m.call_count == 1
+111
View File
@@ -0,0 +1,111 @@
"""The tests for the Home Assistant HTTP component."""
import requests
from homeassistant import bootstrap, const
import homeassistant.components.http as http
from tests.common import get_test_instance_port, get_test_home_assistant
API_PASSWORD = 'test1234'
SERVER_PORT = get_test_instance_port()
HTTP_BASE = '127.0.0.1:{}'.format(SERVER_PORT)
HTTP_BASE_URL = 'http://{}'.format(HTTP_BASE)
HA_HEADERS = {
const.HTTP_HEADER_HA_AUTH: API_PASSWORD,
const.HTTP_HEADER_CONTENT_TYPE: const.CONTENT_TYPE_JSON,
}
CORS_ORIGINS = [HTTP_BASE_URL, HTTP_BASE]
hass = None
def _url(path=''):
"""Helper method to generate URLs."""
return HTTP_BASE_URL + path
# pylint: disable=invalid-name
def setUpModule():
"""Initialize a Home Assistant server."""
global hass
hass = get_test_home_assistant()
bootstrap.setup_component(
hass, http.DOMAIN, {
http.DOMAIN: {
http.CONF_API_PASSWORD: API_PASSWORD,
http.CONF_SERVER_PORT: SERVER_PORT,
http.CONF_CORS_ORIGINS: CORS_ORIGINS,
}
}
)
bootstrap.setup_component(hass, 'api')
hass.start()
# pylint: disable=invalid-name
def tearDownModule():
"""Stop the Home Assistant server."""
hass.stop()
class TestHttp:
"""Test HTTP component."""
def test_cors_allowed_with_password_in_url(self):
"""Test cross origin resource sharing with password in url."""
req = requests.get(_url(const.URL_API),
params={'api_password': API_PASSWORD},
headers={const.HTTP_HEADER_ORIGIN: HTTP_BASE_URL})
allow_origin = const.HTTP_HEADER_ACCESS_CONTROL_ALLOW_ORIGIN
assert req.status_code == 200
assert req.headers.get(allow_origin) == HTTP_BASE_URL
def test_cors_allowed_with_password_in_header(self):
"""Test cross origin resource sharing with password in header."""
headers = {
const.HTTP_HEADER_HA_AUTH: API_PASSWORD,
const.HTTP_HEADER_ORIGIN: HTTP_BASE_URL
}
req = requests.get(_url(const.URL_API), headers=headers)
allow_origin = const.HTTP_HEADER_ACCESS_CONTROL_ALLOW_ORIGIN
assert req.status_code == 200
assert req.headers.get(allow_origin) == HTTP_BASE_URL
def test_cors_denied_without_origin_header(self):
"""Test cross origin resource sharing with password in header."""
headers = {
const.HTTP_HEADER_HA_AUTH: API_PASSWORD
}
req = requests.get(_url(const.URL_API), headers=headers)
allow_origin = const.HTTP_HEADER_ACCESS_CONTROL_ALLOW_ORIGIN
allow_headers = const.HTTP_HEADER_ACCESS_CONTROL_ALLOW_HEADERS
assert req.status_code == 200
assert allow_origin not in req.headers
assert allow_headers not in req.headers
def test_cors_preflight_allowed(self):
"""Test cross origin resource sharing preflight (OPTIONS) request."""
headers = {
const.HTTP_HEADER_ORIGIN: HTTP_BASE_URL,
'Access-Control-Request-Method': 'GET',
'Access-Control-Request-Headers': 'x-ha-access'
}
req = requests.options(_url(const.URL_API), headers=headers)
allow_origin = const.HTTP_HEADER_ACCESS_CONTROL_ALLOW_ORIGIN
allow_headers = const.HTTP_HEADER_ACCESS_CONTROL_ALLOW_HEADERS
assert req.status_code == 200
assert req.headers.get(allow_origin) == HTTP_BASE_URL
assert req.headers.get(allow_headers) == \
const.HTTP_HEADER_HA_AUTH.upper()
+16 -9
View File
@@ -3,10 +3,10 @@ import asyncio
import json
from unittest.mock import patch, MagicMock, mock_open
from aiohttp import web
from homeassistant.components.notify import html5
from tests.common import mock_http_component_app
SUBSCRIPTION_1 = {
'browser': 'chrome',
'subscription': {
@@ -121,7 +121,8 @@ class TestHtml5Notify(object):
assert view.json_path == hass.config.path.return_value
assert view.registrations == {}
app = web.Application(loop=loop)
hass.loop = loop
app = mock_http_component_app(hass)
view.register(app.router)
client = yield from test_client(app)
hass.http.is_banned_ip.return_value = False
@@ -153,7 +154,8 @@ class TestHtml5Notify(object):
view = hass.mock_calls[1][1][0]
app = web.Application(loop=loop)
hass.loop = loop
app = mock_http_component_app(hass)
view.register(app.router)
client = yield from test_client(app)
hass.http.is_banned_ip.return_value = False
@@ -208,7 +210,8 @@ class TestHtml5Notify(object):
assert view.json_path == hass.config.path.return_value
assert view.registrations == config
app = web.Application(loop=loop)
hass.loop = loop
app = mock_http_component_app(hass)
view.register(app.router)
client = yield from test_client(app)
hass.http.is_banned_ip.return_value = False
@@ -253,7 +256,8 @@ class TestHtml5Notify(object):
assert view.json_path == hass.config.path.return_value
assert view.registrations == config
app = web.Application(loop=loop)
hass.loop = loop
app = mock_http_component_app(hass)
view.register(app.router)
client = yield from test_client(app)
hass.http.is_banned_ip.return_value = False
@@ -296,7 +300,8 @@ class TestHtml5Notify(object):
assert view.json_path == hass.config.path.return_value
assert view.registrations == config
app = web.Application(loop=loop)
hass.loop = loop
app = mock_http_component_app(hass)
view.register(app.router)
client = yield from test_client(app)
hass.http.is_banned_ip.return_value = False
@@ -331,7 +336,8 @@ class TestHtml5Notify(object):
view = hass.mock_calls[2][1][0]
app = web.Application(loop=loop)
hass.loop = loop
app = mock_http_component_app(hass)
view.register(app.router)
client = yield from test_client(app)
hass.http.is_banned_ip.return_value = False
@@ -387,7 +393,8 @@ class TestHtml5Notify(object):
bearer_token = "Bearer {}".format(push_payload['data']['jwt'])
app = web.Application(loop=loop)
hass.loop = loop
app = mock_http_component_app(hass)
view.register(app.router)
client = yield from test_client(app)
hass.http.is_banned_ip.return_value = False
+1 -2
View File
@@ -6,7 +6,7 @@ import unittest
import requests
import homeassistant.bootstrap as bootstrap
from homeassistant.components import frontend, http
from homeassistant.components import http
from homeassistant.const import HTTP_HEADER_HA_AUTH
from tests.common import get_test_instance_port, get_test_home_assistant
@@ -45,7 +45,6 @@ def setUpModule():
def tearDownModule():
"""Stop everything that was started."""
hass.stop()
frontend.PANELS = {}
class TestFrontend(unittest.TestCase):
-285
View File
@@ -1,285 +0,0 @@
"""The tests for the Home Assistant HTTP component."""
# pylint: disable=protected-access
import logging
from ipaddress import ip_network
from unittest.mock import patch, mock_open
import requests
from homeassistant import bootstrap, const
import homeassistant.components.http as http
from tests.common import get_test_instance_port, get_test_home_assistant
API_PASSWORD = 'test1234'
SERVER_PORT = get_test_instance_port()
HTTP_BASE = '127.0.0.1:{}'.format(SERVER_PORT)
HTTP_BASE_URL = 'http://{}'.format(HTTP_BASE)
HA_HEADERS = {
const.HTTP_HEADER_HA_AUTH: API_PASSWORD,
const.HTTP_HEADER_CONTENT_TYPE: const.CONTENT_TYPE_JSON,
}
# Don't add 127.0.0.1/::1 as trusted, as it may interfere with other test cases
TRUSTED_NETWORKS = ['192.0.2.0/24', '2001:DB8:ABCD::/48', '100.64.0.1',
'FD01:DB8::1']
TRUSTED_ADDRESSES = ['100.64.0.1', '192.0.2.100', 'FD01:DB8::1',
'2001:DB8:ABCD::1']
UNTRUSTED_ADDRESSES = ['198.51.100.1', '2001:DB8:FA1::1', '127.0.0.1', '::1']
BANNED_IPS = ['200.201.202.203', '100.64.0.1']
CORS_ORIGINS = [HTTP_BASE_URL, HTTP_BASE]
hass = None
def _url(path=''):
"""Helper method to generate URLs."""
return HTTP_BASE_URL + path
# pylint: disable=invalid-name
def setUpModule():
"""Initialize a Home Assistant server."""
global hass
hass = get_test_home_assistant()
hass.bus.listen('test_event', lambda _: _)
hass.states.set('test.test', 'a_state')
bootstrap.setup_component(
hass, http.DOMAIN, {
http.DOMAIN: {
http.CONF_API_PASSWORD: API_PASSWORD,
http.CONF_SERVER_PORT: SERVER_PORT,
http.CONF_CORS_ORIGINS: CORS_ORIGINS,
}
}
)
bootstrap.setup_component(hass, 'api')
hass.http.trusted_networks = [
ip_network(trusted_network)
for trusted_network in TRUSTED_NETWORKS]
hass.http.ip_bans = [http.IpBan(banned_ip)
for banned_ip in BANNED_IPS]
hass.start()
# pylint: disable=invalid-name
def tearDownModule():
"""Stop the Home Assistant server."""
hass.stop()
class TestHttp:
"""Test HTTP component."""
def test_access_denied_without_password(self):
"""Test access without password."""
req = requests.get(_url(const.URL_API))
assert req.status_code == 401
def test_access_denied_with_wrong_password_in_header(self):
"""Test access with wrong password."""
req = requests.get(
_url(const.URL_API),
headers={const.HTTP_HEADER_HA_AUTH: 'wrongpassword'})
assert req.status_code == 401
def test_access_denied_with_x_forwarded_for(self, caplog):
"""Test access denied through the X-Forwarded-For http header."""
hass.http.use_x_forwarded_for = True
for remote_addr in UNTRUSTED_ADDRESSES:
req = requests.get(_url(const.URL_API), headers={
const.HTTP_HEADER_X_FORWARDED_FOR: remote_addr})
assert req.status_code == 401, \
"{} shouldn't be trusted".format(remote_addr)
def test_access_denied_with_untrusted_ip(self, caplog):
"""Test access with an untrusted ip address."""
for remote_addr in UNTRUSTED_ADDRESSES:
with patch('homeassistant.components.http.'
'HomeAssistantWSGI.get_real_ip',
return_value=remote_addr):
req = requests.get(
_url(const.URL_API), params={'api_password': ''})
assert req.status_code == 401, \
"{} shouldn't be trusted".format(remote_addr)
def test_access_with_password_in_header(self, caplog):
"""Test access with password in URL."""
# Hide logging from requests package that we use to test logging
caplog.set_level(
logging.WARNING, logger='requests.packages.urllib3.connectionpool')
req = requests.get(
_url(const.URL_API),
headers={const.HTTP_HEADER_HA_AUTH: API_PASSWORD})
assert req.status_code == 200
logs = caplog.text
# assert const.URL_API in logs
assert API_PASSWORD not in logs
def test_access_denied_with_wrong_password_in_url(self):
"""Test access with wrong password."""
req = requests.get(
_url(const.URL_API), params={'api_password': 'wrongpassword'})
assert req.status_code == 401
def test_access_with_password_in_url(self, caplog):
"""Test access with password in URL."""
# Hide logging from requests package that we use to test logging
caplog.set_level(
logging.WARNING, logger='requests.packages.urllib3.connectionpool')
req = requests.get(
_url(const.URL_API), params={'api_password': API_PASSWORD})
assert req.status_code == 200
logs = caplog.text
# assert const.URL_API in logs
assert API_PASSWORD not in logs
def test_access_granted_with_x_forwarded_for(self, caplog):
"""Test access denied through the X-Forwarded-For http header."""
hass.http.use_x_forwarded_for = True
for remote_addr in TRUSTED_ADDRESSES:
req = requests.get(_url(const.URL_API), headers={
const.HTTP_HEADER_X_FORWARDED_FOR: remote_addr})
assert req.status_code == 200, \
"{} should be trusted".format(remote_addr)
def test_access_granted_with_trusted_ip(self, caplog):
"""Test access with trusted addresses."""
for remote_addr in TRUSTED_ADDRESSES:
with patch('homeassistant.components.http.'
'HomeAssistantWSGI.get_real_ip',
return_value=remote_addr):
req = requests.get(
_url(const.URL_API), params={'api_password': ''})
assert req.status_code == 200, \
'{} should be trusted'.format(remote_addr)
def test_cors_allowed_with_password_in_url(self):
"""Test cross origin resource sharing with password in url."""
req = requests.get(_url(const.URL_API),
params={'api_password': API_PASSWORD},
headers={const.HTTP_HEADER_ORIGIN: HTTP_BASE_URL})
allow_origin = const.HTTP_HEADER_ACCESS_CONTROL_ALLOW_ORIGIN
assert req.status_code == 200
assert req.headers.get(allow_origin) == HTTP_BASE_URL
def test_cors_allowed_with_password_in_header(self):
"""Test cross origin resource sharing with password in header."""
headers = {
const.HTTP_HEADER_HA_AUTH: API_PASSWORD,
const.HTTP_HEADER_ORIGIN: HTTP_BASE_URL
}
req = requests.get(_url(const.URL_API), headers=headers)
allow_origin = const.HTTP_HEADER_ACCESS_CONTROL_ALLOW_ORIGIN
assert req.status_code == 200
assert req.headers.get(allow_origin) == HTTP_BASE_URL
def test_cors_denied_without_origin_header(self):
"""Test cross origin resource sharing with password in header."""
headers = {
const.HTTP_HEADER_HA_AUTH: API_PASSWORD
}
req = requests.get(_url(const.URL_API), headers=headers)
allow_origin = const.HTTP_HEADER_ACCESS_CONTROL_ALLOW_ORIGIN
allow_headers = const.HTTP_HEADER_ACCESS_CONTROL_ALLOW_HEADERS
assert req.status_code == 200
assert allow_origin not in req.headers
assert allow_headers not in req.headers
def test_cors_preflight_allowed(self):
"""Test cross origin resource sharing preflight (OPTIONS) request."""
headers = {
const.HTTP_HEADER_ORIGIN: HTTP_BASE_URL,
'Access-Control-Request-Method': 'GET',
'Access-Control-Request-Headers': 'x-ha-access'
}
req = requests.options(_url(const.URL_API), headers=headers)
allow_origin = const.HTTP_HEADER_ACCESS_CONTROL_ALLOW_ORIGIN
allow_headers = const.HTTP_HEADER_ACCESS_CONTROL_ALLOW_HEADERS
assert req.status_code == 200
assert req.headers.get(allow_origin) == HTTP_BASE_URL
assert req.headers.get(allow_headers) == \
const.HTTP_HEADER_HA_AUTH.upper()
def test_access_from_banned_ip(self):
"""Test accessing to server from banned IP. Both trusted and not."""
hass.http.is_ban_enabled = True
for remote_addr in BANNED_IPS:
with patch('homeassistant.components.http.'
'HomeAssistantWSGI.get_real_ip',
return_value=remote_addr):
req = requests.get(
_url(const.URL_API))
assert req.status_code == 403
def test_access_from_banned_ip_when_ban_is_off(self):
"""Test accessing to server from banned IP when feature is off"""
hass.http.is_ban_enabled = False
for remote_addr in BANNED_IPS:
with patch('homeassistant.components.http.'
'HomeAssistantWSGI.get_real_ip',
return_value=remote_addr):
req = requests.get(
_url(const.URL_API),
headers={const.HTTP_HEADER_HA_AUTH: API_PASSWORD})
assert req.status_code == 200
def test_ip_bans_file_creation(self):
"""Testing if banned IP file created"""
hass.http.is_ban_enabled = True
hass.http.login_threshold = 1
m = mock_open()
def call_server():
with patch('homeassistant.components.http.'
'HomeAssistantWSGI.get_real_ip',
return_value="200.201.202.204"):
return requests.get(
_url(const.URL_API),
headers={const.HTTP_HEADER_HA_AUTH: 'Wrong password'})
with patch('homeassistant.components.http.open', m, create=True):
req = call_server()
assert req.status_code == 401
assert len(hass.http.ip_bans) == len(BANNED_IPS)
assert m.call_count == 0
req = call_server()
assert req.status_code == 401
assert len(hass.http.ip_bans) == len(BANNED_IPS) + 1
m.assert_called_once_with(hass.config.path(http.IP_BANS), 'a')
req = call_server()
assert req.status_code == 403
assert m.call_count == 1
+8
View File
@@ -165,7 +165,15 @@ class TestCheckConfig(unittest.TestCase):
self.assertDictEqual({
'components': {'http': {'api_password': 'abc123',
'cors_allowed_origins': [],
'development': '0',
'ip_ban_enabled': True,
'login_attempts_threshold': -1,
'server_host': '0.0.0.0',
'server_port': 8123,
'ssl_certificate': None,
'ssl_key': None,
'trusted_networks': [],
'use_x_forwarded_for': False}},
'except': {},
'secret_cache': {secrets_path: {'http_pw': 'abc123'}},