http: reimplement X-Forwarded-For parsing (#4355)

This feature needs to be enabled through the `http.use_x_forwarded_for` option,
satisfying security concerns of spoofed remote addresses in untrusted network
environments.

The testsuite was enhanced to explicitly test the functionality of the
header.

Fixes #4265.

Signed-off-by: Martin Weinelt <hexa@darmstadt.ccc.de>

homeassistant/components/emulated_hue.py

@@ -76,6 +76,7 @@ def setup(hass, yaml_config):
         ssl_certificate=None,
         ssl_key=None,
         cors_origins=[],
+        use_x_forwarded_for=False,
         trusted_networks=[]
     )
 

homeassistant/components/http.py

@@ -28,7 +28,7 @@
 from homeassistant.const import (
     SERVER_PORT, HTTP_HEADER_HA_AUTH,  # HTTP_HEADER_CACHE_CONTROL,
     CONTENT_TYPE_JSON, ALLOWED_CORS_HEADERS, EVENT_HOMEASSISTANT_STOP,
-    EVENT_HOMEASSISTANT_START)
+    EVENT_HOMEASSISTANT_START, HTTP_HEADER_X_FORWARDED_FOR)
 import homeassistant.helpers.config_validation as cv
 from homeassistant.components import persistent_notification
 
@@ -42,6 +42,7 @@
 CONF_SSL_CERTIFICATE = 'ssl_certificate'
 CONF_SSL_KEY = 'ssl_key'
 CONF_CORS_ORIGINS = 'cors_allowed_origins'
+CONF_USE_X_FORWARDED_FOR = 'use_x_forwarded_for'
 CONF_TRUSTED_NETWORKS = 'trusted_networks'
 
 DATA_API_PASSWORD = 'api_password'
@@ -82,6 +83,7 @@
         vol.Optional(CONF_SSL_CERTIFICATE): cv.isfile,
         vol.Optional(CONF_SSL_KEY): cv.isfile,
         vol.Optional(CONF_CORS_ORIGINS): vol.All(cv.ensure_list, [cv.string]),
+        vol.Optional(CONF_USE_X_FORWARDED_FOR, default=False): cv.boolean,
         vol.Optional(CONF_TRUSTED_NETWORKS):
             vol.All(cv.ensure_list, [ip_network])
     }),
@@ -125,6 +127,7 @@ def setup(hass, config):
     ssl_certificate = conf.get(CONF_SSL_CERTIFICATE)
     ssl_key = conf.get(CONF_SSL_KEY)
     cors_origins = conf.get(CONF_CORS_ORIGINS, [])
+    use_x_forwarded_for = conf.get(CONF_USE_X_FORWARDED_FOR, False)
     trusted_networks = [
         ip_network(trusted_network)
         for trusted_network in conf.get(CONF_TRUSTED_NETWORKS, [])]
@@ -138,6 +141,7 @@ def setup(hass, config):
         ssl_certificate=ssl_certificate,
         ssl_key=ssl_key,
         cors_origins=cors_origins,
+        use_x_forwarded_for=use_x_forwarded_for,
         trusted_networks=trusted_networks
     )
 
@@ -248,7 +252,7 @@ class HomeAssistantWSGI(object):
 
     def __init__(self, hass, development, api_password, ssl_certificate,
                  ssl_key, server_host, server_port, cors_origins,
-                 trusted_networks):
+                 use_x_forwarded_for, trusted_networks):
         """Initialize the WSGI Home Assistant server."""
         import aiohttp_cors
 
@@ -260,6 +264,7 @@ def __init__(self, hass, development, api_password, ssl_certificate,
         self.ssl_key = ssl_key
         self.server_host = server_host
         self.server_port = server_port
+        self.use_x_forwarded_for = use_x_forwarded_for
         self.trusted_networks = trusted_networks
         self.event_forwarder = None
         self._handler = None
@@ -366,11 +371,15 @@ def stop(self):
         yield from self._handler.finish_connections(60.0)
         yield from self.app.cleanup()
 
-    @staticmethod
-    def get_real_ip(request):
+    def get_real_ip(self, request):
         """Return the clients correct ip address, even in proxied setups."""
-        peername = request.transport.get_extra_info('peername')
-        return peername[0] if peername is not None else None
+        if self.use_x_forwarded_for \
+                and HTTP_HEADER_X_FORWARDED_FOR in request.headers:
+            return request.headers.get(
+                HTTP_HEADER_X_FORWARDED_FOR).split(',')[0]
+        else:
+            peername = request.transport.get_extra_info('peername')
+            return peername[0] if peername is not None else None
 
     def is_trusted_ip(self, remote_addr):
         """Match an ip address against trusted CIDR networks."""
@@ -452,7 +461,7 @@ def request_handler_factory(view, handler):
     @asyncio.coroutine
     def handle(request):
         """Handle incoming request."""
-        remote_addr = HomeAssistantWSGI.get_real_ip(request)
+        remote_addr = view.hass.http.get_real_ip(request)
 
         # Auth code verbose on purpose
         authenticated = False

homeassistant/const.py

@@ -360,6 +360,7 @@
 HTTP_HEADER_CACHE_CONTROL = 'Cache-Control'
 HTTP_HEADER_EXPIRES = 'Expires'
 HTTP_HEADER_ORIGIN = 'Origin'
+HTTP_HEADER_X_FORWARDED_FOR = 'X-Forwarded-For'
 HTTP_HEADER_X_REQUESTED_WITH = 'X-Requested-With'
 HTTP_HEADER_ACCEPT = 'Accept'
 HTTP_HEADER_ACCESS_CONTROL_ALLOW_ORIGIN = 'Access-Control-Allow-Origin'

tests/components/test_http.py

@@ -22,6 +22,10 @@
 # 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']
+
 
 CORS_ORIGINS = [HTTP_BASE_URL, HTTP_BASE]
 
@@ -85,10 +89,19 @@ def test_access_denied_with_wrong_password_in_header(self):
 
         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 ['198.51.100.1', '2001:DB8:FA1::1', '127.0.0.1',
-                            '::1']:
+        for remote_addr in UNTRUSTED_ADDRESSES:
             with patch('homeassistant.components.http.'
                        'HomeAssistantWSGI.get_real_ip',
                        return_value=remote_addr):
@@ -138,10 +151,19 @@ def test_access_with_password_in_url(self, caplog):
         # assert const.URL_API in logs
         assert API_PASSWORD not in logs
 
-    def test_access_with_trusted_ip(self, caplog):
+    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 ['100.64.0.1', '192.0.2.100', 'FD01:DB8::1',
-                            '2001:DB8:ABCD::1']:
+        for remote_addr in TRUSTED_ADDRESSES:
             with patch('homeassistant.components.http.'
                        'HomeAssistantWSGI.get_real_ip',
                        return_value=remote_addr):

tests/scripts/test_check_config.py

@@ -165,7 +165,8 @@ def test_secrets(self):
 
             self.assertDictEqual({
                 'components': {'http': {'api_password': 'abc123',
-                                        'server_port': 8123}},
+                                        'server_port': 8123,
+                                        'use_x_forwarded_for': False}},
                 'except': {},
                 'secret_cache': {secrets_path: {'http_pw': 'abc123'}},
                 'secrets': {'http_pw': 'abc123'},