Advanced Ip filtering

homeassistant/components/emulated_hue.py

@@ -77,7 +77,10 @@ def setup(hass, yaml_config):
         ssl_key=None,
         cors_origins=[],
         use_x_forwarded_for=False,
-        trusted_networks=[]
+        trusted_networks=[],
+        ip_bans=[],
+        login_threshold=0,
+        is_ban_enabled=False
     )
 
     server.register_view(DescriptionXmlView(hass, config))

homeassistant/components/http.py

@@ -5,32 +5,36 @@
 https://home-assistant.io/components/http/
 """
 import asyncio
-import hmac
 import json
 import logging
 import mimetypes
-import os
-from pathlib import Path
-import re
 import ssl
+from datetime import datetime
 from ipaddress import ip_address, ip_network
+from pathlib import Path
 
+import hmac
+import os
+import re
 import voluptuous as vol
 from aiohttp import web, hdrs
 from aiohttp.file_sender import FileSender
 from aiohttp.web_exceptions import (
-    HTTPUnauthorized, HTTPMovedPermanently, HTTPNotModified)
+    HTTPUnauthorized, HTTPMovedPermanently, HTTPNotModified, HTTPForbidden)
 from aiohttp.web_urldispatcher import StaticRoute
 
-from homeassistant.core import is_callback
+import homeassistant.helpers.config_validation as cv
 import homeassistant.remote as rem
 from homeassistant import util
+from homeassistant.components import persistent_notification
+from homeassistant.config import load_yaml_config_file
 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, HTTP_HEADER_X_FORWARDED_FOR)
-import homeassistant.helpers.config_validation as cv
-from homeassistant.components import persistent_notification
+from homeassistant.core import is_callback
+from homeassistant.exceptions import HomeAssistantError
+from homeassistant.util.yaml import dump
 
 DOMAIN = 'http'
 REQUIREMENTS = ('aiohttp_cors==0.4.0',)
@@ -44,9 +48,16 @@
 CONF_CORS_ORIGINS = 'cors_allowed_origins'
 CONF_USE_X_FORWARDED_FOR = 'use_x_forwarded_for'
 CONF_TRUSTED_NETWORKS = 'trusted_networks'
+CONF_LOGIN_ATTEMPTS_THRESHOLD = 'login_attempts_threshold'
+CONF_IP_BAN_ENABLED = 'ip_ban_enabled'
 
 DATA_API_PASSWORD = 'api_password'
 NOTIFICATION_ID_LOGIN = 'http-login'
+NOTIFICATION_ID_BAN = 'ip-ban'
+
+IP_BANS = 'ip_bans.yaml'
+ATTR_BANNED_AT = "banned_at"
+
 
 # TLS configuation follows the best-practice guidelines specified here:
 # https://wiki.mozilla.org/Security/Server_Side_TLS
@@ -85,7 +96,9 @@
         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])
+            vol.All(cv.ensure_list, [ip_network]),
+        vol.Optional(CONF_LOGIN_ATTEMPTS_THRESHOLD): cv.positive_int,
+        vol.Optional(CONF_IP_BAN_ENABLED): cv.boolean
     }),
 }, extra=vol.ALLOW_EXTRA)
 
@@ -131,6 +144,9 @@ def setup(hass, config):
     trusted_networks = [
         ip_network(trusted_network)
         for trusted_network in conf.get(CONF_TRUSTED_NETWORKS, [])]
+    is_ban_enabled = bool(conf.get(CONF_IP_BAN_ENABLED, False))
+    login_threshold = int(conf.get(CONF_LOGIN_ATTEMPTS_THRESHOLD, -1))
+    ip_bans = load_ip_bans_config(hass.config.path(IP_BANS))
 
     server = HomeAssistantWSGI(
         hass,
@@ -142,7 +158,10 @@ def setup(hass, config):
         ssl_key=ssl_key,
         cors_origins=cors_origins,
         use_x_forwarded_for=use_x_forwarded_for,
-        trusted_networks=trusted_networks
+        trusted_networks=trusted_networks,
+        ip_bans=ip_bans,
+        login_threshold=login_threshold,
+        is_ban_enabled=is_ban_enabled
     )
 
     @asyncio.coroutine
@@ -252,7 +271,8 @@ class HomeAssistantWSGI(object):
 
     def __init__(self, hass, development, api_password, ssl_certificate,
                  ssl_key, server_host, server_port, cors_origins,
-                 use_x_forwarded_for, trusted_networks):
+                 use_x_forwarded_for, trusted_networks,
+                 ip_bans, login_threshold, is_ban_enabled):
         """Initialize the WSGI Home Assistant server."""
         import aiohttp_cors
 
@@ -269,6 +289,10 @@ def __init__(self, hass, development, api_password, ssl_certificate,
         self.event_forwarder = None
         self._handler = None
         self.server = None
+        self.login_threshold = login_threshold
+        self.ip_bans = ip_bans if ip_bans is not None else []
+        self.failed_login_attempts = {}
+        self.is_ban_enabled = is_ban_enabled
 
         if cors_origins:
             self.cors = aiohttp_cors.setup(self.app, defaults={
@@ -386,6 +410,38 @@ def is_trusted_ip(self, remote_addr):
         return any(ip_address(remote_addr) in trusted_network
                    for trusted_network in self.hass.http.trusted_networks)
 
+    def wrong_login_attempt(self, remote_addr):
+        """Registering wrong login attempt."""
+        if not self.is_trusted_ip(remote_addr) \
+                and self.login_threshold > 0 and self.is_ban_enabled:
+            if remote_addr in self.failed_login_attempts:
+                self.failed_login_attempts[remote_addr] += 1
+            else:
+                self.failed_login_attempts[remote_addr] = 1
+
+            if self.failed_login_attempts[remote_addr] > self.login_threshold:
+                new_ban = IpBan(remote_addr)
+                self.ip_bans.append(new_ban)
+                update_ip_bans_config(self.hass.config.path(IP_BANS), new_ban)
+                _LOGGER.warning('Banned IP %s for too many login attempts',
+                                remote_addr)
+                persistent_notification.async_create(
+                    self.hass,
+                    'To many login attempts from {}'.format(remote_addr),
+                    'Banning IP address', NOTIFICATION_ID_BAN)
+
+    def is_banned_ip(self, remote_addr):
+        """Check if IP address is in a ban list."""
+        if not self.is_ban_enabled:
+            return False
+
+        ip_address_ = ip_address(remote_addr)
+        for ip_ban in self.ip_bans:
+            if ip_ban.ip_address == ip_address_:
+                return True
+
+        return False
+
 
 class HomeAssistantView(object):
     """Base view for all views."""
@@ -466,6 +522,9 @@ def handle(request):
 
         remote_addr = view.hass.http.get_real_ip(request)
 
+        if view.hass.http.is_banned_ip(remote_addr):
+            raise HTTPForbidden()
+
         # Auth code verbose on purpose
         authenticated = False
 
@@ -485,6 +544,7 @@ def handle(request):
             authenticated = True
 
         if view.requires_auth and not authenticated:
+            view.hass.http.wrong_login_attempt(remote_addr)
             _LOGGER.warning('Login attempt or request with an invalid '
                             'password from %s', remote_addr)
             persistent_notification.async_create(
@@ -526,3 +586,47 @@ def handle(request):
         return web.Response(body=result, status=status_code)
 
     return handle
+
+
+class IpBan(object):
+    """Represents banned IP address."""
+
+    def __init__(self, ip_ban: str, banned_at: datetime=None) -> None:
+        """Initializing Ip Ban object."""
+        self.ip_address = ip_address(ip_ban)
+        self.banned_at = banned_at
+        if self.banned_at is None:
+            self.banned_at = datetime.utcnow()
+
+
+def load_ip_bans_config(path: str):
+    """Loading list of banned IPs from config file."""
+    ip_list = []
+    ip_schema = vol.Schema({
+        vol.Optional('banned_at'): vol.Any(None, cv.exact_time)
+    })
+
+    try:
+        list_ = load_yaml_config_file(path)
+        for ip_ban in list_:
+            try:
+                ban = ip_schema(list_[ip_ban])
+                ban['ip_ban'] = ip_address(ip_ban)
+                ip_list.append(IpBan(**ban))
+            except vol.Invalid:
+                continue
+
+    except(HomeAssistantError, FileNotFoundError):
+        return []
+
+    return ip_list
+
+
+def update_ip_bans_config(path: str, ip_ban: IpBan):
+    """Update config file with new banned IP address."""
+    with open(path, 'a') as out:
+        ip_ = {str(ip_ban.ip_address): {
+            ATTR_BANNED_AT: ip_ban.banned_at.strftime("%Y-%m-%dT%H:%M:%S")
+        }}
+        out.write('\n')
+        out.write(dump(ip_))

homeassistant/helpers/config_validation.py

@@ -1,6 +1,6 @@
 """Helpers for config validation using voluptuous."""
 from collections import OrderedDict
-from datetime import timedelta
+from datetime import timedelta, datetime, date
 import os
 import re
 from urllib.parse import urlparse
@@ -297,6 +297,13 @@ def time(value):
     return time_val
 
 
+def exact_time(value):
+    """Validate timestamp."""
+    if isinstance(value, date) or isinstance(value, datetime):
+        return value
+    return datetime.strptime(value, "%Y-%m-%dT%H:%M:%S")
+
+
 def time_zone(value):
     """Validate timezone."""
     if dt_util.get_time_zone(value) is not None:

tests/components/notify/test_html5.py

@@ -124,6 +124,7 @@ def test_registering_new_device_view(self, loop, test_client):
             app = web.Application(loop=loop)
             view.register(app.router)
             client = yield from test_client(app)
+            hass.http.is_banned_ip.return_value = False
             resp = yield from client.post(REGISTER_URL,
                                           data=json.dumps(SUBSCRIPTION_1))
 
@@ -155,6 +156,7 @@ def test_registering_new_device_validation(self, loop, test_client):
             app = web.Application(loop=loop)
             view.register(app.router)
             client = yield from test_client(app)
+            hass.http.is_banned_ip.return_value = False
 
             resp = yield from client.post(REGISTER_URL, data=json.dumps({
                 'browser': 'invalid browser',
@@ -209,6 +211,7 @@ def test_unregistering_device_view(self, loop, test_client):
             app = web.Application(loop=loop)
             view.register(app.router)
             client = yield from test_client(app)
+            hass.http.is_banned_ip.return_value = False
 
             resp = yield from client.delete(REGISTER_URL, data=json.dumps({
                 'subscription': SUBSCRIPTION_1['subscription'],
@@ -253,6 +256,7 @@ def test_unregister_device_view_handle_unknown_subscription(self, loop,
             app = web.Application(loop=loop)
             view.register(app.router)
             client = yield from test_client(app)
+            hass.http.is_banned_ip.return_value = False
 
             resp = yield from client.delete(REGISTER_URL, data=json.dumps({
                 'subscription': SUBSCRIPTION_3['subscription']
@@ -295,6 +299,7 @@ def test_unregistering_device_view_handles_json_safe_error(self, loop,
             app = web.Application(loop=loop)
             view.register(app.router)
             client = yield from test_client(app)
+            hass.http.is_banned_ip.return_value = False
 
             with patch('homeassistant.components.notify.html5._save_config',
                        return_value=False):
@@ -329,6 +334,7 @@ def test_callback_view_no_jwt(self, loop, test_client):
             app = web.Application(loop=loop)
             view.register(app.router)
             client = yield from test_client(app)
+            hass.http.is_banned_ip.return_value = False
 
             resp = yield from client.post(PUBLISH_URL, data=json.dumps({
                 'type': 'push',
@@ -384,6 +390,7 @@ def test_callback_view_with_jwt(self, loop, test_client):
             app = web.Application(loop=loop)
             view.register(app.router)
             client = yield from test_client(app)
+            hass.http.is_banned_ip.return_value = False
 
             resp = yield from client.post(PUBLISH_URL, data=json.dumps({
                 'type': 'push',

tests/components/test_http.py

@@ -2,7 +2,7 @@
 # pylint: disable=protected-access
 import logging
 from ipaddress import ip_network
-from unittest.mock import patch
+from unittest.mock import patch, mock_open
 
 import requests
 
@@ -25,7 +25,7 @@
 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]
 
@@ -63,6 +63,9 @@ def setUpModule():
         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()
 
 
@@ -227,3 +230,56 @@ def test_cors_preflight_allowed(self):
         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