Add ip_ban_whitelist for HTTP Integration

Currently in the HTTP integration, you are able to set ip_ban_enabled, this is great… except the iOS Companion app seems to occasionaly fail to authenicate for some reason, and this causes my devices (phones / tablets) to stop working until I realise, after which I need to login via a device which has hopefully not been banned and clear the IP’s listed in the ip_ban.yaml.

Ideally, I’d like to whitelist my local subnet to ensure that my local devices are never locked out, even if they for some reason do an authentication failure (thanks iOS companion app). The whitelist whould allow you to add individual IP’s or IP subnets, much in the same way as you can with the trusted proxies.

An example configuration is below with the (to be added) ip_ban_whitelist:

# Example configuration.yaml entry
http:
  server_port: 8123
  ssl_certificate: /ssl/fullchain.pem
  ssl_key: /ssl/privkey.pem
  use_x_forwarded_for: true
  trusted_proxies:
    - 172.30.32.0/24
  ip_ban_enabled: true
  login_attempts_threshold: 3
  ip_ban_whitelist:
    - 192.168.1.0/24

Excellent idea!

I’ve been running into a similar situation lately where my PC tries to re-login to the HA server and can’t (also, like you, for an unknown reason - I really have no idea why). So I need to do as you and clear the IP ban and when logging back in it works fine with literally nothing being changed except clearing the ban.

Have there been any updates to this or alternative solutions found? This issue effectively makes ip_ban usage really painful. Several times a week I’m getting locked out and have to purge the ip_bans file…

This is a great idea, as my Android companion app also sometimes causes an entry in the IP ban list.

Unless they implement this, ip_ban is completely broken. I have switched it off after three days, because my router somehow caused authentication errors. When your router is banned, no device in your LAN works lol.

The real solution is: devices running HA Companion app should not create false positives.

There are some improvements but still… if your ban treshold is quite low, devices will still get blocked… :frowning:

I rebooted PI from SSH, refreshed the existing loaded browser page and banned myself without any password prompt visible so will need a workaround.
I am thinking of creating a background script that checks for ip_bans.yaml every 5min and delete the file if ip is 192.* and restart ha

Edit:

Here is the script that run as sudo on startup

#!/bin/bash
ipban="/usr/share/hassio/homeassistant/ip_bans.yaml"
while true; do
	sleep 5m
	if [ -f $ipban ]; then
		if grep -q 192.168.0 "$ipban"; then
			#replaces 192.168.0. to 1.2.3.
			sed -i 's/192\.168\.0\./1.2.3./' $ipban
			docker restart homeassistant
		fi
	fi
done

Added a line in /var/spool/cron/crontabs/root
@reboot /usr/share/hassio/homeassistant/resetban.sh

1 Like

Just make sure that external IPs does not log as internal/router IP.
In my case it’s always the router that gets banned, and i believe that is because the router routes it and the real IP is not forwarded.
I know there is a solution for it but in my case I would rather have whole HA go in to lockdown mode if someone actually tries than it banning one IP at the time.

Just remove: login_attempts_threshold: 3 from your configuration.yaml file and it stops banning entirely… (it will give a notification)

Excellent idea!

I wanted to write a request myself, and I see you brought me forward.
I really hope it is implemented.

This is awesome and exactly what I’m looking for!
I’ve installed a first Fire HD tablet in Kiosk mode in my home and it was banned out of nothing half an hour ago … so I had to disable the banlist (which is not the best idea from the point of security)

I’m trying a practical solution to promote it, I have no experience with Python.
I tried to play around a bit with the file: homeassistant/components/http/init.py

"""Support to serve the Home Assistant API as WSGI application."""
from __future__ import annotations

from ipaddress import ip_network
import logging
import os
import ssl
from typing import Any, Final, Optional, TypedDict, cast

from aiohttp import web
from aiohttp.typedefs import StrOrURL
from aiohttp.web_exceptions import HTTPMovedPermanently, HTTPRedirection
import voluptuous as vol

from homeassistant.components.network import async_get_source_ip
from homeassistant.const import EVENT_HOMEASSISTANT_STOP, SERVER_PORT
from homeassistant.core import Event, HomeAssistant
from homeassistant.helpers import storage
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.typing import ConfigType
from homeassistant.loader import bind_hass
from homeassistant.setup import async_start_setup, async_when_setup_or_start
from homeassistant.util import ssl as ssl_util

from .auth import setup_auth
from .ban import setup_bans
from .const import KEY_AUTHENTICATED, KEY_HASS, KEY_HASS_USER  # noqa: F401
from .cors import setup_cors
from .forwarded import async_setup_forwarded
from .request_context import current_request, setup_request_context
from .security_filter import setup_security_filter
from .static import CACHE_HEADERS, CachingStaticResource
from .view import HomeAssistantView
from .web_runner import HomeAssistantTCPSite

DOMAIN: Final = "http"

CONF_SERVER_HOST: Final = "server_host"
CONF_SERVER_PORT: Final = "server_port"
CONF_BASE_URL: Final = "base_url"
CONF_SSL_CERTIFICATE: Final = "ssl_certificate"
CONF_SSL_PEER_CERTIFICATE: Final = "ssl_peer_certificate"
CONF_SSL_KEY: Final = "ssl_key"
CONF_CORS_ORIGINS: Final = "cors_allowed_origins"
CONF_USE_X_FORWARDED_FOR: Final = "use_x_forwarded_for"
CONF_TRUSTED_PROXIES: Final = "trusted_proxies"
CONF_LOGIN_ATTEMPTS_THRESHOLD: Final = "login_attempts_threshold"
CONF_IP_BAN_ENABLED: Final = "ip_ban_enabled"
CONF_IP_BAN_WHITELIST: Final = "ip_ban_whitelist"
CONF_SSL_PROFILE: Final = "ssl_profile"

SSL_MODERN: Final = "modern"
SSL_INTERMEDIATE: Final = "intermediate"

_LOGGER: Final = logging.getLogger(__name__)

DEFAULT_DEVELOPMENT: Final = "0"
# Cast to be able to load custom cards.
# My to be able to check url and version info.
DEFAULT_CORS: Final[list[str]] = ["https://cast.home-assistant.io"]
NO_LOGIN_ATTEMPT_THRESHOLD: Final = -1

MAX_CLIENT_SIZE: Final = 1024 ** 2 * 16

STORAGE_KEY: Final = DOMAIN
STORAGE_VERSION: Final = 1
SAVE_DELAY: Final = 180

HTTP_SCHEMA: Final = vol.All(
    cv.deprecated(CONF_BASE_URL),
    vol.Schema(
        {
            vol.Optional(CONF_SERVER_HOST): vol.All(
                cv.ensure_list, vol.Length(min=1), [cv.string]
            ),
            vol.Optional(CONF_SERVER_PORT, default=SERVER_PORT): cv.port,
            vol.Optional(CONF_BASE_URL): cv.string,
            vol.Optional(CONF_SSL_CERTIFICATE): cv.isfile,
            vol.Optional(CONF_SSL_PEER_CERTIFICATE): cv.isfile,
            vol.Optional(CONF_SSL_KEY): cv.isfile,
            vol.Optional(CONF_CORS_ORIGINS, default=DEFAULT_CORS): vol.All(
                cv.ensure_list, [cv.string]
            ),
            vol.Inclusive(CONF_USE_X_FORWARDED_FOR, "proxy"): cv.boolean,
            vol.Inclusive(CONF_TRUSTED_PROXIES, "proxy"): vol.All(
                cv.ensure_list, [ip_network]
            ),
            vol.Optional(
                CONF_LOGIN_ATTEMPTS_THRESHOLD, default=NO_LOGIN_ATTEMPT_THRESHOLD
            ): vol.Any(cv.positive_int, NO_LOGIN_ATTEMPT_THRESHOLD),
            vol.Inclusive(CONF_IP_BAN_ENABLED, "list"): cv.boolean,
			vol.Inclusive(CONF_IP_BAN_WHITELIST, "list"): vol.All(
                cv.ensure_list, [ip_network]
            ),
            vol.Optional(CONF_SSL_PROFILE, default=SSL_MODERN): vol.In(
                [SSL_INTERMEDIATE, SSL_MODERN]
            ),
        }
    ),
)

CONFIG_SCHEMA: Final = vol.Schema({DOMAIN: HTTP_SCHEMA}, extra=vol.ALLOW_EXTRA)


class ConfData(TypedDict, total=False):
    """Typed dict for config data."""

    server_host: list[str]
    server_port: int
    base_url: str
    ssl_certificate: str
    ssl_peer_certificate: str
    ssl_key: str
    cors_allowed_origins: list[str]
    use_x_forwarded_for: bool
    trusted_proxies: list[str]
    login_attempts_threshold: int
    ip_ban_enabled: bool
	ip_ban_whitelist: list[str]
    ssl_profile: str


@bind_hass
async def async_get_last_config(hass: HomeAssistant) -> dict | None:
    """Return the last known working config."""
    store = storage.Store(hass, STORAGE_VERSION, STORAGE_KEY)
    return cast(Optional[dict], await store.async_load())


class ApiConfig:
    """Configuration settings for API server."""

    def __init__(
        self,
        local_ip: str,
        host: str,
        port: int,
        use_ssl: bool,
    ) -> None:
        """Initialize a new API config object."""
        self.local_ip = local_ip
        self.host = host
        self.port = port
        self.use_ssl = use_ssl


async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
    """Set up the HTTP API and debug interface."""
    conf: ConfData | None = config.get(DOMAIN)

    if conf is None:
        conf = cast(ConfData, HTTP_SCHEMA({}))

    server_host = conf.get(CONF_SERVER_HOST)
    server_port = conf[CONF_SERVER_PORT]
    ssl_certificate = conf.get(CONF_SSL_CERTIFICATE)
    ssl_peer_certificate = conf.get(CONF_SSL_PEER_CERTIFICATE)
    ssl_key = conf.get(CONF_SSL_KEY)
    cors_origins = conf[CONF_CORS_ORIGINS]
    use_x_forwarded_for = conf.get(CONF_USE_X_FORWARDED_FOR, False)
    trusted_proxies = conf.get(CONF_TRUSTED_PROXIES) or []
    is_ban_enabled = conf[CONF_IP_BAN_ENABLED]
	ip_whitelist = conf.get(CONF_IP_BAN_WHITELIST) or []
    login_threshold = conf[CONF_LOGIN_ATTEMPTS_THRESHOLD]
    ssl_profile = conf[CONF_SSL_PROFILE]

    server = HomeAssistantHTTP(
        hass,
        server_host=server_host,
        server_port=server_port,
        ssl_certificate=ssl_certificate,
        ssl_peer_certificate=ssl_peer_certificate,
        ssl_key=ssl_key,
        cors_origins=cors_origins,
        use_x_forwarded_for=use_x_forwarded_for,
        trusted_proxies=trusted_proxies,
        login_threshold=login_threshold,
        is_ban_enabled=is_ban_enabled,
		ip_whitelist=ip_whitelist,
        ssl_profile=ssl_profile,
    )

    async def stop_server(event: Event) -> None:
        """Stop the server."""
        await server.stop()

    async def start_server(*_: Any) -> None:
        """Start the server."""
        with async_start_setup(hass, ["http"]):
            hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, stop_server)
            # We already checked it's not None.
            assert conf is not None
            await start_http_server_and_save_config(hass, dict(conf), server)

    async_when_setup_or_start(hass, "frontend", start_server)

    hass.http = server

    local_ip = await async_get_source_ip(hass)

    host = local_ip
    if server_host is not None:
        # Assume the first server host name provided as API host
        host = server_host[0]

    hass.config.api = ApiConfig(
        local_ip, host, server_port, ssl_certificate is not None
    )

    return True


class HomeAssistantHTTP:
    """HTTP server for Home Assistant."""

    def __init__(
        self,
        hass: HomeAssistant,
        ssl_certificate: str | None,
        ssl_peer_certificate: str | None,
        ssl_key: str | None,
        server_host: list[str] | None,
        server_port: int,
        cors_origins: list[str],
        use_x_forwarded_for: bool,
        trusted_proxies: list[str],
        login_threshold: int,
        is_ban_enabled: bool,
		ip_whitelist: list[str],
        ssl_profile: str,
    ) -> None:
        """Initialize the HTTP Home Assistant server."""
        app = self.app = web.Application(
            middlewares=[], client_max_size=MAX_CLIENT_SIZE
        )
        app[KEY_HASS] = hass

        # Order matters, security filters middle ware needs to go first,
        # forwarded middleware needs to go second.
        setup_security_filter(app)

        async_setup_forwarded(app, use_x_forwarded_for, trusted_proxies)

        setup_request_context(app, current_request)

        if is_ban_enabled:
            setup_bans(hass, app, ip_whitelist, login_threshold)

        setup_auth(hass, app)

        setup_cors(app, cors_origins)

        self.hass = hass
        self.ssl_certificate = ssl_certificate
        self.ssl_peer_certificate = ssl_peer_certificate
        self.ssl_key = ssl_key
        self.server_host = server_host
        self.server_port = server_port
        self.trusted_proxies = trusted_proxies
        self.is_ban_enabled = is_ban_enabled
		self.ip_whitelist = ip_whitelist
        self.ssl_profile = ssl_profile
        self._handler = None
        self.runner: web.AppRunner | None = None
        self.site: HomeAssistantTCPSite | None = None

    def register_view(self, view: HomeAssistantView) -> None:
        """Register a view with the WSGI server.

        The view argument must be a class that inherits from HomeAssistantView.
        It is optional to instantiate it before registering; this method will
        handle it either way.
        """
        if isinstance(view, type):
            # Instantiate the view, if needed
            view = view()

        if not hasattr(view, "url"):
            class_name = view.__class__.__name__
            raise AttributeError(f'{class_name} missing required attribute "url"')

        if not hasattr(view, "name"):
            class_name = view.__class__.__name__
            raise AttributeError(f'{class_name} missing required attribute "name"')

        view.register(self.app, self.app.router)

    def register_redirect(
        self,
        url: str,
        redirect_to: StrOrURL,
        *,
        redirect_exc: type[HTTPRedirection] = HTTPMovedPermanently,
    ) -> None:
        """Register a redirect with the server.

        If given this must be either a string or callable. In case of a
        callable it's called with the url adapter that triggered the match and
        the values of the URL as keyword arguments and has to return the target
        for the redirect, otherwise it has to be a string with placeholders in
        rule syntax.
        """

        async def redirect(request: web.Request) -> web.StreamResponse:
            """Redirect to location."""
            # Should be instance of aiohttp.web_exceptions._HTTPMove.
            raise redirect_exc(redirect_to)  # type: ignore[arg-type,misc]

        self.app["allow_configured_cors"](
            self.app.router.add_route("GET", url, redirect)
        )

    def register_static_path(
        self, url_path: str, path: str, cache_headers: bool = True
    ) -> None:
        """Register a folder or file to serve as a static path."""
        if os.path.isdir(path):
            if cache_headers:
                resource: CachingStaticResource | web.StaticResource = (
                    CachingStaticResource(url_path, path)
                )
            else:
                resource = web.StaticResource(url_path, path)
            self.app.router.register_resource(resource)
            self.app["allow_configured_cors"](resource)
            return

        async def serve_file(request: web.Request) -> web.FileResponse:
            """Serve file from disk."""
            if cache_headers:
                return web.FileResponse(path, headers=CACHE_HEADERS)
            return web.FileResponse(path)

        self.app["allow_configured_cors"](
            self.app.router.add_route("GET", url_path, serve_file)
        )

    async def start(self) -> None:
        """Start the aiohttp server."""
        context: ssl.SSLContext | None
        if self.ssl_certificate:
            try:
                if self.ssl_profile == SSL_INTERMEDIATE:
                    context = ssl_util.server_context_intermediate()
                else:
                    context = ssl_util.server_context_modern()
                await self.hass.async_add_executor_job(
                    context.load_cert_chain, self.ssl_certificate, self.ssl_key
                )
            except OSError as error:
                _LOGGER.error(
                    "Could not read SSL certificate from %s: %s",
                    self.ssl_certificate,
                    error,
                )
                return

            if self.ssl_peer_certificate:
                context.verify_mode = ssl.CERT_REQUIRED
                await self.hass.async_add_executor_job(
                    context.load_verify_locations, self.ssl_peer_certificate
                )

        else:
            context = None

        # Aiohttp freezes apps after start so that no changes can be made.
        # However in Home Assistant components can be discovered after boot.
        # This will now raise a RunTimeError.
        # To work around this we now prevent the router from getting frozen
        # pylint: disable=protected-access
        self.app._router.freeze = lambda: None  # type: ignore[assignment]

        self.runner = web.AppRunner(self.app)
        await self.runner.setup()

        self.site = HomeAssistantTCPSite(
            self.runner, self.server_host, self.server_port, ssl_context=context
        )
        try:
            await self.site.start()
        except OSError as error:
            _LOGGER.error(
                "Failed to create HTTP server at port %d: %s", self.server_port, error
            )

        _LOGGER.info("Now listening on port %d", self.server_port)

    async def stop(self) -> None:
        """Stop the aiohttp server."""
        if self.site is not None:
            await self.site.stop()
        if self.runner is not None:
            await self.runner.cleanup()


async def start_http_server_and_save_config(
    hass: HomeAssistant, conf: dict, server: HomeAssistantHTTP
) -> None:
    """Startup the http server and save the config."""
    await server.start()

    # If we are set up successful, we store the HTTP settings for safe mode.
    store = storage.Store(hass, STORAGE_VERSION, STORAGE_KEY)

    if CONF_TRUSTED_PROXIES in conf:
        conf[CONF_TRUSTED_PROXIES] = [
            str(ip.network_address) for ip in conf[CONF_TRUSTED_PROXIES]
        ]

    store.async_delay_save(lambda: conf, SAVE_DELAY)

Can anyone help with thet?
So that PR can be submitted to expedite it

1 Like

Nice work already @shilo
Can you share the link to your forked version of homeassistant/components/http/init.py so that we can see what changes you made.

Link in the next post

@niro1987

You changed line 91 from __init__.py

vol.Inclusive(CONF_IP_BAN_ENABLED, default=True): cv.boolean,

To

vol.Inclusive(CONF_IP_BAN_ENABLED, "list"): cv.boolean,

I think you need to revert that back.

Have you tested this in a dev environment?

Excellent idea !
I had connection problems because of my android companion app and now is my router IP banned. The only solution left is to set ip_ban_enabled to false, which problematic from the point of security.

How can I whitelist my router ?

If your phone made the router banned then any outside attack will also be shown as router.

You’re right. I have to find a way to solve this.